keystone: write_capability: Check if report was already transmitted

core/capabilities/targets/write_target.go

@@ -10,6 +10,7 @@ import (
 	chainselectors "github.com/smartcontractkit/chain-selectors"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	commontypes "github.com/smartcontractkit/chainlink-common/pkg/types"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
 	"github.com/smartcontractkit/chainlink-common/pkg/values"
 	txmgrcommon "github.com/smartcontractkit/chainlink/v2/common/txmgr"
@@ -18,6 +19,8 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/chains/legacyevm"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/keystone/generated/forwarder"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	evm "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm"
+	relayevmtypes "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/types"
 )
 
 var forwardABI = evmtypes.MustGetABI(forwarder.KeystoneForwarderMetaData.ABI)
@@ -84,9 +87,21 @@ func parseConfig(rawConfig *values.Map) (EvmConfig, error) {
 	return config, err
 }
 
+func success() <-chan capabilities.CapabilityResponse {
+	callback := make(chan capabilities.CapabilityResponse)
+	go func() {
+		// TODO: cast tx.Error to Err (or Value to Value?)
+		callback <- capabilities.CapabilityResponse{
+			Value: nil,
+			Err:   nil,
+		}
+		close(callback)
+	}()
+	return callback
+}
+
 func (cap *EvmWrite) Execute(ctx context.Context, request capabilities.CapabilityRequest) (<-chan capabilities.CapabilityResponse, error) {
 	cap.lggr.Debugw("Execute", "request", request)
-	// TODO: idempotency
 
 	txm := cap.chain.TxManager()
 
@@ -96,7 +111,6 @@ func (cap *EvmWrite) Execute(ctx context.Context, request capabilities.Capabilit
 	if err != nil {
 		return nil, err
 	}
-
 	var inputs struct {
 		Report     []byte
 		Signatures [][]byte
@@ -108,20 +122,48 @@ func (cap *EvmWrite) Execute(ctx context.Context, request capabilities.Capabilit
 	if inputs.Report == nil {
 		// We received any empty report -- this means we should skip transmission.
 		cap.lggr.Debugw("Skipping empty report", "request", request)
-		callback := make(chan capabilities.CapabilityResponse)
-		go func() {
-			// TODO: cast tx.Error to Err (or Value to Value?)
-			callback <- capabilities.CapabilityResponse{
-				Value: nil,
-				Err:   nil,
-			}
-			close(callback)
-		}()
-		return callback, nil
+		return success(), nil
 	}
 
 	// TODO: validate encoded report is prefixed with workflowID and executionID that match the request meta
 
+	// Check whether value was already transmitted on chain
+	cr, err := evm.NewChainReaderService(ctx, cap.lggr, cap.chain.LogPoller(), cap.chain.Client(), relayevmtypes.ChainReaderConfig{
+		Contracts: map[string]relayevmtypes.ChainContractReader{
+			"forwarder": {
+				ContractABI: forwarder.KeystoneForwarderABI,
+				Configs: map[string]*relayevmtypes.ChainReaderDefinition{
+					"getTransmitter": {
+						ChainSpecificName: "getTransmitter",
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	var transmitter common.Address
+	cr.Bind(ctx, []commontypes.BoundContract{{
+		Address: config.ForwarderAddress().String(),
+		Name:    "forwarder",
+		// Pending: false, // ???
+	}})
+	queryInputs := struct {
+		Receiver            string
+		WorkflowExecutionID []byte
+	}{
+		Receiver:            reqConfig.Address,
+		WorkflowExecutionID: []byte(request.Metadata.WorkflowExecutionID),
+	}
+	if err := cr.GetLatestValue(ctx, "forwarder", "getTransmitter", queryInputs, &transmitter); err != nil {
+		return nil, err
+	}
+	if transmitter != common.HexToAddress("0x0") {
+		// report already transmitted, early return
+		return success(), nil
+	}
+
 	// construct forwarder payload
 	calldata, err := forwardABI.Pack("report", common.HexToAddress(reqConfig.Address), inputs.Report, inputs.Signatures)
 	if err != nil {
@@ -132,37 +174,24 @@ func (cap *EvmWrite) Execute(ctx context.Context, request capabilities.Capabilit
 		// FwdrDestAddress could also be set for better logging but it's used for various purposes around Operator Forwarders
 		WorkflowExecutionID: &request.Metadata.WorkflowExecutionID,
 	}
-	strategy := txmgrcommon.NewSendEveryStrategy()
-
-	checker := txmgr.TransmitCheckerSpec{
-		CheckerType: txmgr.TransmitCheckerTypeSimulate,
-	}
 	req := txmgr.TxRequest{
 		FromAddress:    config.FromAddress().Address(),
 		ToAddress:      config.ForwarderAddress().Address(),
 		EncodedPayload: calldata,
 		FeeLimit:       uint64(defaultGasLimit),
 		Meta:           txMeta,
-		Strategy:       strategy,
-		Checker:        checker,
+		Strategy:       txmgrcommon.NewSendEveryStrategy(),
+		Checker: txmgr.TransmitCheckerSpec{
+			CheckerType: txmgr.TransmitCheckerTypeSimulate,
+		},
 		// SignalCallback:   true, TODO: add code that checks if a workflow id is present, if so, route callback to chainwriter rather than pipeline
 	}
 	tx, err := txm.CreateTransaction(ctx, req)
 	if err != nil {
 		return nil, err
 	}
 	cap.lggr.Debugw("Transaction submitted", "request", request, "transaction", tx)
-
-	callback := make(chan capabilities.CapabilityResponse)
-	go func() {
-		// TODO: cast tx.Error to Err (or Value to Value?)
-		callback <- capabilities.CapabilityResponse{
-			Value: nil,
-			Err:   nil,
-		}
-		close(callback)
-	}()
-	return callback, nil
+	return success(), nil
 }
 
 func (cap *EvmWrite) RegisterToWorkflow(ctx context.Context, request capabilities.RegisterToWorkflowRequest) error {

core/capabilities/targets/write_target_test.go

@@ -1,9 +1,11 @@
 package targets_test
 
 import (
+	"encoding/hex"
 	"math/big"
 	"testing"
 
+	"github.com/ethereum/go-ethereum"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/values"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/targets"
@@ -24,12 +26,21 @@ import (
 
 var forwardABI = types.MustGetABI(forwarder.KeystoneForwarderMetaData.ABI)
 
+func addr(t *testing.T, lastByte string) []byte {
+	contractAddr, err := hex.DecodeString("00000000000000000000000000000000000000000000000000000000000000" + lastByte)
+	require.NoError(t, err)
+	return contractAddr
+}
+
 func TestEvmWrite(t *testing.T) {
 	chain := evmmocks.NewChain(t)
 
 	txManager := txmmocks.NewMockEvmTxManager(t)
 	chain.On("ID").Return(big.NewInt(11155111))
 	chain.On("TxManager").Return(txManager)
+	chain.On("LogPoller").Return(nil)
+	ethClient := evmtest.NewEthClientMockWithDefaultChain(t)
+	chain.On("Client").Return(ethClient)
 
 	cfg := configtest.NewGeneralConfig(t, func(c *chainlink.Config, s *chainlink.Secrets) {
 		a := testutils.NewAddress()
@@ -48,7 +59,9 @@ func TestEvmWrite(t *testing.T) {
 	capability := targets.NewEvmWrite(chain, logger.TestLogger(t))
 	ctx := testutils.Context(t)
 
-	config, err := values.NewMap(map[string]any{})
+	config, err := values.NewMap(map[string]any{
+		"address": "0xDeaDbeefdEAdbeefdEadbEEFdeadbeEFdEaDbeeF",
+	})
 	require.NoError(t, err)
 
 	inputs, err := values.NewMap(map[string]any{
@@ -65,6 +78,13 @@ func TestEvmWrite(t *testing.T) {
 		Inputs: inputs,
 	}
 
+	ethClient.On("CallContract", mock.Anything, mock.IsType(ethereum.CallMsg{}), mock.IsType(&big.Int{})).Run(func(args mock.Arguments) {
+		// callMsg := args.Get(1).(ethereum.CallMsg)
+		// assert.Equal(t, rollups.ArbGasInfoAddress, callMsg.To.String())
+		// assert.Equal(t, rollups.ArbGasInfo_getPricesInArbGas, fmt.Sprintf("%x", callMsg.Data))
+		// assert.Equal(t, big.NewInt(-1), blockNumber)
+	}).Return(addr(t, "00"), nil)
+
 	txManager.On("CreateTransaction", mock.Anything, mock.Anything).Return(txmgr.Tx{}, nil).Run(func(args mock.Arguments) {
 		req := args.Get(1).(txmgr.TxRequest)
 		payload := make(map[string]any)
@@ -107,8 +127,7 @@ func TestEvmWrite_EmptyReport(t *testing.T) {
 	ctx := testutils.Context(t)
 
 	config, err := values.NewMap(map[string]any{
-		"abi":    "receive(report bytes)",
-		"params": []any{"$(report)"},
+		"address": "0xDeaDbeefdEAdbeefdEadbEEFdeadbeEFdEaDbeeF",
 	})
 	require.NoError(t, err)