remote capabilities errors are properly handled as caperrors (#21746)

* remote capabilities errors are properly handled as caperrors

* lint

* Update core/capabilities/remote/executable/request/client_request.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: fernandezlautaro

core/capabilities/remote/executable/request/client_request.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-protos/workflows/go/events"
@@ -27,6 +28,32 @@ import (
 	p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types"
 )
 
+// errRemoteCapabilityExecuteError preserves the legacy "TRANSPORT : ErrorMsg" string from the
+// remote executable client while wrapping a deserialized caperrors.Error so callers can
+// errors.As into caperrors.Error after RPC (see capability_executor metrics).
+type errRemoteCapabilityExecuteError struct {
+	s    string
+	wrap caperrors.Error
+}
+
+func (e *errRemoteCapabilityExecuteError) Error() string { return e.s }
+
+func (e *errRemoteCapabilityExecuteError) Unwrap() error { return e.wrap }
+
+func newRemoteCapabilityExecuteError(transportErr types.Error, errMsg string) error {
+	return &errRemoteCapabilityExecuteError{
+		s:    fmt.Sprintf("%s : %s", transportErr, errMsg),
+		wrap: caperrors.DeserializeErrorFromString(errMsg),
+	}
+}
+
+func newRemoteCapabilityExecuteErrorWithMessage(display string, errMsg string) error {
+	return &errRemoteCapabilityExecuteError{
+		s:    display,
+		wrap: caperrors.DeserializeErrorFromString(errMsg),
+	}
+}
+
 type clientResponse struct {
 	Result []byte
 	Err    error
@@ -351,9 +378,12 @@ func (c *ClientRequest) OnMessage(_ context.Context, msg *types.MessageBody) err
 		}
 
 		if c.errorCount[msg.ErrorMsg] == c.requiredIdenticalResponses {
-			c.sendResponse(clientResponse{Err: fmt.Errorf("%s : %s", msg.Error, msg.ErrorMsg)})
+			c.sendResponse(clientResponse{Err: newRemoteCapabilityExecuteError(msg.Error, msg.ErrorMsg)})
 		} else if c.totalErrorCount == c.remoteNodeCount-c.requiredIdenticalResponses+1 {
-			c.sendResponse(clientResponse{Err: fmt.Errorf("received %d errors, last error %s : %s", c.totalErrorCount, msg.Error, msg.ErrorMsg)})
+			c.sendResponse(clientResponse{Err: newRemoteCapabilityExecuteErrorWithMessage(
+				fmt.Sprintf("received %d errors, last error %s : %s", c.totalErrorCount, msg.Error, msg.ErrorMsg),
+				msg.ErrorMsg,
+			)})
 		}
 	}
 	return nil

core/capabilities/remote/executable/request/client_request_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder/beholdertest"
 	commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb"
 	"github.com/smartcontractkit/chainlink-protos/cre/go/values"
 	"github.com/smartcontractkit/chainlink-protos/workflows/go/events"
@@ -229,6 +230,58 @@ func Test_ClientRequest_MessageValidation(t *testing.T) {
 		response := <-req.ResponseChan()
 
 		assert.Equal(t, fmt.Sprintf("%s : %s", types.Error_INTERNAL_ERROR, assert.AnError.Error()), response.Err.Error())
+
+		var capErr caperrors.Error
+		require.ErrorAs(t, response.Err, &capErr)
+		assert.Equal(t, caperrors.OriginSystem, capErr.Origin(), "non-serialized ErrorMsg falls back to private system capability error")
+		assert.Equal(t, caperrors.VisibilityPrivate, capErr.Visibility())
+		assert.Equal(t, caperrors.Unknown, capErr.Code())
+	})
+
+	t.Run("Error response with serialized caperrors unwraps correctly as usererror", func(t *testing.T) {
+		ctx := t.Context()
+		capabilityPeers, capDonInfo, capInfo := capabilityDon(t, 4, 1)
+
+		dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)}
+		req, err := request.NewClientExecuteRequest(ctx, logger.Test(t), capabilityRequest, capInfo,
+			workflowDonInfo, dispatcher, 10*time.Minute, nil, "")
+		require.NoError(t, err)
+		defer req.Cancel(errors.New("test end"))
+
+		<-dispatcher.msgs
+		<-dispatcher.msgs
+		assert.Empty(t, dispatcher.msgs)
+
+		serialized := caperrors.NewPublicUserError(errors.New("rpc error: EVM error invalid argument"), caperrors.FailedPrecondition).SerializeToRemoteString()
+		msgWithError := &types.MessageBody{
+			CapabilityId:    capInfo.ID,
+			CapabilityDonId: capDonInfo.ID,
+			CallerDonId:     workflowDonInfo.ID,
+			Method:          types.MethodExecute,
+			Payload:         rawResponse,
+			MessageId:       []byte("messageID"),
+			Error:           types.Error_INTERNAL_ERROR,
+			ErrorMsg:        serialized,
+		}
+
+		msgWithError.Sender = capabilityPeers[0][:]
+		err = req.OnMessage(ctx, msgWithError)
+		require.NoError(t, err)
+
+		msgWithError.Sender = capabilityPeers[1][:]
+		err = req.OnMessage(ctx, msgWithError)
+		require.NoError(t, err)
+
+		response := <-req.ResponseChan()
+
+		wantDisplay := fmt.Sprintf("%s : %s", types.Error_INTERNAL_ERROR, serialized)
+		assert.Equal(t, wantDisplay, response.Err.Error(), "It should be equal to 'Public:User:FailedPrecondition:rpc error: EVM error invalid argument'")
+
+		var capErr caperrors.Error
+		require.ErrorAs(t, response.Err, &capErr)
+		assert.Equal(t, caperrors.OriginUser, capErr.Origin())
+		assert.Equal(t, caperrors.VisibilityPublic, capErr.Visibility())
+		assert.Equal(t, caperrors.FailedPrecondition, capErr.Code())
 	})
 
 	t.Run("Send three messages with different errors", func(t *testing.T) {

core/services/workflows/v2/capability_executor.go

@@ -226,7 +226,8 @@ func (c *ExecutionHelper) callCapability(ctx context.Context, request *sdkpb.Cap
 
 		execLogger.Debugw("Capability execution failed", "err", err)
 		_ = events.EmitCapabilityFinishedEvent(ctx, loggerLabels, c.WorkflowExecutionID, request.Id, meteringRef, store.StatusErrored, request.Method, err)
-		c.metrics.With(platform.KeyCapabilityID, request.Id, platform.KeyCapabilityErrorCode, caperrors.Unknown.String()).IncrementCapabilityFailureCounter(ctx)
+		// TODO shouldn't all capabilities *always* return a typed error, and if so shouldn't the following metric alert us there's a bug we need to fix?
+		c.metrics.With(platform.KeyCapabilityID, request.Id, platform.KeyCapabilityErrorCode, "BUG").IncrementCapabilityFailureCounter(ctx)
 		c.metrics.IncrementTotalWorkflowStepErrorsCounter(ctx)
 		return nil, fmt.Errorf("failed to execute capability: %w", err)
 	}