Update Nexus error model to use Temporal failures

chasm/lib/callback/chasm_invocation.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/base64"
 	"fmt"
-	"io"
 
 	"github.com/google/uuid"
 	"github.com/nexus-rpc/sdk-go/nexus"
@@ -28,7 +27,7 @@ import (
 type chasmInvocation struct {
 	nexus      *callbackspb.Callback_Nexus
 	attempt    int32
-	completion nexusrpc.OperationCompletion
+	completion nexusrpc.CompleteOperationOptions
 	requestID  string
 }
 
@@ -135,47 +134,44 @@ func (c chasmInvocation) getHistoryRequest(
 		RequestId:    c.requestID,
 	}
 
-	switch op := c.completion.(type) {
-	case *nexusrpc.OperationCompletionSuccessful:
-		payloadBody, err := io.ReadAll(op.Reader)
-		if err != nil {
-			return nil, fmt.Errorf("failed to read payload: %v", err)
-		}
-
+	if c.completion.Error == nil {
 		var payload *commonpb.Payload
-		if payloadBody != nil {
-			content := &nexus.Content{
-				Header: op.Reader.Header,
-				Data:   payloadBody,
-			}
-			err := commonnexus.PayloadSerializer.Deserialize(content, &payload)
-			if err != nil {
-				return nil, fmt.Errorf("failed to deserialize payload: %v", err)
+		if c.completion.Result != nil {
+			var ok bool
+			payload, ok = c.completion.Result.(*commonpb.Payload)
+			if !ok {
+				return nil, fmt.Errorf("invalid result, expected a payload, got: %T", c.completion.Result)
 			}
 		}
 
 		req = &historyservice.CompleteNexusOperationChasmRequest{
 			Outcome: &historyservice.CompleteNexusOperationChasmRequest_Success{
 				Success: payload,
 			},
-			CloseTime:  timestamppb.New(op.CloseTime),
+			CloseTime:  timestamppb.New(c.completion.CloseTime),
 			Completion: completion,
 		}
-	case *nexusrpc.OperationCompletionUnsuccessful:
-		apiFailure, err := commonnexus.NexusFailureToAPIFailure(op.Failure, true)
+	} else {
+		failure, err := nexusrpc.DefaultFailureConverter().ErrorToFailure(c.completion.Error)
 		if err != nil {
-			return nil, fmt.Errorf("failed to convert failure type: %v", err)
+			return nil, fmt.Errorf("failed to convert error to failure: %w", err)
+		}
+		// Unwrap the operation error since it's not meant to be sent for Temporal->Temporal completions.
+		if failure.Cause != nil {
+			failure = *failure.Cause
+		}
+		apiFailure, err := commonnexus.NexusFailureToTemporalFailure(failure)
+		if err != nil {
+			return nil, fmt.Errorf("failed to convert failure type: %w", err)
 		}
 
 		req = &historyservice.CompleteNexusOperationChasmRequest{
 			Completion: completion,
 			Outcome: &historyservice.CompleteNexusOperationChasmRequest_Failure{
 				Failure: apiFailure,
 			},
-			CloseTime: timestamppb.New(op.CloseTime),
+			CloseTime: timestamppb.New(c.completion.CloseTime),
 		}
-	default:
-		return nil, fmt.Errorf("unexpected nexus.OperationCompletion: %v", completion)
 	}
 
 	return req, nil

chasm/lib/callback/component.go

@@ -15,7 +15,7 @@ import (
 )
 
 type CompletionSource interface {
-	GetNexusCompletion(ctx chasm.Context, requestID string) (nexusrpc.OperationCompletion, error)
+	GetNexusCompletion(ctx chasm.Context, requestID string) (nexusrpc.CompleteOperationOptions, error)
 }
 
 var _ chasm.Component = (*Callback)(nil)

chasm/lib/callback/executors_test.go

@@ -40,13 +40,13 @@ type mockNexusCompletionGetterComponent struct {
 
 	Empty *emptypb.Empty
 
-	completion nexusrpc.OperationCompletion
+	completion nexusrpc.CompleteOperationOptions
 	err        error
 
 	Callback chasm.Field[*Callback]
 }
 
-func (m *mockNexusCompletionGetterComponent) GetNexusCompletion(_ chasm.Context, requestID string) (nexusrpc.OperationCompletion, error) {
+func (m *mockNexusCompletionGetterComponent) GetNexusCompletion(_ chasm.Context, requestID string) (nexusrpc.CompleteOperationOptions, error) {
 	return m.completion, m.err
 }
 
@@ -81,7 +81,7 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 			caller: func(r *http.Request) (*http.Response, error) {
 				return &http.Response{StatusCode: 200, Body: http.NoBody}, nil
 			},
-			expectedMetricOutcome: "status:200",
+			expectedMetricOutcome: "success",
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
 				require.NoError(t, err)
 				require.Equal(t, callbackspb.CALLBACK_STATUS_SUCCEEDED, cb.Status)
@@ -104,7 +104,7 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 			caller: func(r *http.Request) (*http.Response, error) {
 				return &http.Response{StatusCode: 500, Body: http.NoBody}, nil
 			},
-			expectedMetricOutcome: "status:500",
+			expectedMetricOutcome: "handler-error:INTERNAL",
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
 				var destDownErr *queueserrors.DestinationDownError
 				require.ErrorAs(t, err, &destDownErr)
@@ -116,7 +116,7 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 			caller: func(r *http.Request) (*http.Response, error) {
 				return &http.Response{StatusCode: 400, Body: http.NoBody}, nil
 			},
-			expectedMetricOutcome: "status:400",
+			expectedMetricOutcome: "handler-error:BAD_REQUEST",
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
 				require.NoError(t, err)
 				require.Equal(t, callbackspb.CALLBACK_STATUS_FAILED, cb.Status)
@@ -210,8 +210,7 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 			}
 
 			// Create completion
-			completion, err := nexusrpc.NewOperationCompletionSuccessful(nil, nexusrpc.OperationCompletionSuccessfulOptions{})
-			require.NoError(t, err)
+			completion := nexusrpc.CompleteOperationOptions{}
 
 			// Set up the CompletionSource field to return our mock completion
 			root.SetRootComponent(&mockNexusCompletionGetterComponent{
@@ -371,7 +370,7 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 	cases := []struct {
 		name               string
 		setupHistoryClient func(*testing.T, *gomock.Controller) resource.HistoryClient
-		completion         nexusrpc.OperationCompletion
+		completion         nexusrpc.CompleteOperationOptions
 		headerValue        string
 		assertOutcome      func(*testing.T, *Callback, error)
 	}{
@@ -397,16 +396,11 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				})
 				return client
 			},
-			completion: func() nexusrpc.OperationCompletion {
-				comp, err := nexusrpc.NewOperationCompletionSuccessful(
-					createPayloadBytes([]byte("result-data")),
-					nexusrpc.OperationCompletionSuccessfulOptions{
-						Serializer: commonnexus.PayloadSerializer,
-						CloseTime:  dummyTime,
-					},
-				)
-				require.NoError(t, err)
-				return comp
+			completion: func() nexusrpc.CompleteOperationOptions {
+				return nexusrpc.CompleteOperationOptions{
+					Result:    createPayloadBytes([]byte("result-data")),
+					CloseTime: dummyTime,
+				}
 			}(),
 			headerValue: encodedRef,
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
@@ -430,18 +424,14 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				})
 				return client
 			},
-			completion: func() nexusrpc.OperationCompletion {
-				comp, err := nexusrpc.NewOperationCompletionUnsuccessful(
-					&nexus.OperationError{
+			completion: func() nexusrpc.CompleteOperationOptions {
+				return nexusrpc.CompleteOperationOptions{
+					Error: &nexus.OperationError{
 						State: nexus.OperationStateFailed,
 						Cause: &nexus.FailureError{Failure: nexus.Failure{Message: "operation failed"}},
 					},
-					nexusrpc.OperationCompletionUnsuccessfulOptions{
-						CloseTime: dummyTime,
-					},
-				)
-				require.NoError(t, err)
-				return comp
+					CloseTime: dummyTime,
+				}
 			}(),
 			headerValue: encodedRef,
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
@@ -459,15 +449,10 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				).Return(nil, status.Error(codes.Unavailable, "service unavailable"))
 				return client
 			},
-			completion: func() nexusrpc.OperationCompletion {
-				comp, err := nexusrpc.NewOperationCompletionSuccessful(
-					createPayloadBytes([]byte("result-data")),
-					nexusrpc.OperationCompletionSuccessfulOptions{
-						Serializer: commonnexus.PayloadSerializer,
-					},
-				)
-				require.NoError(t, err)
-				return comp
+			completion: func() nexusrpc.CompleteOperationOptions {
+				return nexusrpc.CompleteOperationOptions{
+					Result: createPayloadBytes([]byte("result-data")),
+				}
 			}(),
 			headerValue: encodedRef,
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
@@ -485,15 +470,10 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				).Return(nil, status.Error(codes.InvalidArgument, "invalid request"))
 				return client
 			},
-			completion: func() nexusrpc.OperationCompletion {
-				comp, err := nexusrpc.NewOperationCompletionSuccessful(
-					createPayloadBytes([]byte("result-data")),
-					nexusrpc.OperationCompletionSuccessfulOptions{
-						Serializer: commonnexus.PayloadSerializer,
-					},
-				)
-				require.NoError(t, err)
-				return comp
+			completion: func() nexusrpc.CompleteOperationOptions {
+				return nexusrpc.CompleteOperationOptions{
+					Result: createPayloadBytes([]byte("result-data")),
+				}
 			}(),
 			headerValue: encodedRef,
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
@@ -507,15 +487,10 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				// No RPC call expected
 				return historyservicemock.NewMockHistoryServiceClient(ctrl)
 			},
-			completion: func() nexusrpc.OperationCompletion {
-				comp, err := nexusrpc.NewOperationCompletionSuccessful(
-					createPayloadBytes([]byte("result-data")),
-					nexusrpc.OperationCompletionSuccessfulOptions{
-						Serializer: commonnexus.PayloadSerializer,
-					},
-				)
-				require.NoError(t, err)
-				return comp
+			completion: func() nexusrpc.CompleteOperationOptions {
+				return nexusrpc.CompleteOperationOptions{
+					Result: createPayloadBytes([]byte("result-data")),
+				}
 			}(),
 			headerValue: "invalid-base64!!!",
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {
@@ -529,15 +504,10 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				// No RPC call expected
 				return historyservicemock.NewMockHistoryServiceClient(ctrl)
 			},
-			completion: func() nexusrpc.OperationCompletion {
-				comp, err := nexusrpc.NewOperationCompletionSuccessful(
-					createPayloadBytes([]byte("result-data")),
-					nexusrpc.OperationCompletionSuccessfulOptions{
-						Serializer: commonnexus.PayloadSerializer,
-					},
-				)
-				require.NoError(t, err)
-				return comp
+			completion: func() nexusrpc.CompleteOperationOptions {
+				return nexusrpc.CompleteOperationOptions{
+					Result: createPayloadBytes([]byte("result-data")),
+				}
 			}(),
 			headerValue: base64.RawURLEncoding.EncodeToString([]byte("not-valid-protobuf")),
 			assertOutcome: func(t *testing.T, cb *Callback, err error) {