Fix error mapping and complete the upgrade

Author: bergundy

internal/error.go

@@ -297,7 +297,11 @@ type (
 		// Operation name.
 		Operation string
 		// Operation ID - may be empty if the operation completed synchronously.
+		//
+		// Deprecated: Use OperationToken instead.
 		OperationID string
+		// Operation token - may be empty if the operation completed synchronously.
+		OperationToken string
 		// Chained cause - typically an ApplicationError or a CanceledError.
 		Cause error
 	}
@@ -909,8 +913,8 @@ func (e *ChildWorkflowExecutionError) RetryState() enumspb.RetryState {
 // Error implements the error interface.
 func (e *NexusOperationError) Error() string {
 	msg := fmt.Sprintf(
-		"%s (endpoint: %q, service: %q, operation: %q, operation ID: %q, scheduledEventID: %d)",
-		e.Message, e.Endpoint, e.Service, e.Operation, e.OperationID, e.ScheduledEventID)
+		"%s (endpoint: %q, service: %q, operation: %q, operation token: %q, scheduledEventID: %d)",
+		e.Message, e.Endpoint, e.Service, e.Operation, e.OperationToken, e.ScheduledEventID)
 	if e.Cause != nil {
 		msg = fmt.Sprintf("%s: %v", msg, e.Cause)
 	}

internal/failure_converter.go

@@ -170,12 +170,17 @@ func (dfc *DefaultFailureConverter) ErrorToFailure(err error) *failurepb.Failure
 		}
 		failure.FailureInfo = &failurepb.Failure_ChildWorkflowExecutionFailureInfo{ChildWorkflowExecutionFailureInfo: failureInfo}
 	case *NexusOperationError:
+		var token = err.OperationToken
+		if token == "" {
+			token = err.OperationID
+		}
 		failureInfo := &failurepb.NexusOperationFailureInfo{
 			ScheduledEventId: err.ScheduledEventID,
 			Endpoint:         err.Endpoint,
 			Service:          err.Service,
 			Operation:        err.Operation,
-			OperationId:      err.OperationID,
+			OperationId:      token,
+			OperationToken:   token,
 		}
 		failure.FailureInfo = &failurepb.Failure_NexusOperationExecutionFailureInfo{NexusOperationExecutionFailureInfo: failureInfo}
 	case *nexus.HandlerError:
@@ -278,6 +283,10 @@ func (dfc *DefaultFailureConverter) FailureToError(failure *failurepb.Failure) e
 			dfc.FailureToError(failure.GetCause()),
 		)
 	} else if info := failure.GetNexusOperationExecutionFailureInfo(); info != nil {
+		token := info.GetOperationToken()
+		if token == "" {
+			token = info.GetOperationId()
+		}
 		err = &NexusOperationError{
 			Message:          failure.Message,
 			Cause:            dfc.FailureToError(failure.GetCause()),
@@ -286,7 +295,8 @@ func (dfc *DefaultFailureConverter) FailureToError(failure *failurepb.Failure) e
 			Endpoint:         info.GetEndpoint(),
 			Service:          info.GetService(),
 			Operation:        info.GetOperation(),
-			OperationID:      info.GetOperationId(),
+			OperationToken:   token,
+			OperationID:      token,
 		}
 	} else if info := failure.GetNexusHandlerFailureInfo(); info != nil {
 		err = &nexus.HandlerError{

internal/interceptor.go

@@ -217,8 +217,8 @@ type RequestCancelNexusOperationInput struct {
 	Client NexusClient
 	// Operation name or OperationReference from the Nexus SDK.
 	Operation any
-	// Operation ID. May be empty if the operation is synchronous or has not started yet.
-	ID string
+	// Operation Token. May be empty if the operation is synchronous or has not started yet.
+	Token string
 	// seq number. For internal use only.
 	seq int64
 }

internal/internal_event_handlers.go

@@ -85,7 +85,7 @@ type (
 	}
 
 	scheduledNexusOperation struct {
-		startedCallback   func(operationID string, err error)
+		startedCallback   func(token string, err error)
 		completedCallback func(result *commonpb.Payload, err error)
 		endpoint          string
 		service           string
@@ -627,7 +627,7 @@ func (wc *workflowEnvironmentImpl) ExecuteChildWorkflow(
 		tagWorkflowType, params.WorkflowType.Name)
 }
 
-func (wc *workflowEnvironmentImpl) ExecuteNexusOperation(params executeNexusOperationParams, callback func(*commonpb.Payload, error), startedHandler func(opID string, e error)) int64 {
+func (wc *workflowEnvironmentImpl) ExecuteNexusOperation(params executeNexusOperationParams, callback func(*commonpb.Payload, error), startedHandler func(token string, e error)) int64 {
 	seq := wc.GenerateSequence()
 	scheduleTaskAttr := &commandpb.ScheduleNexusOperationCommandAttributes{
 		Endpoint:               params.client.Endpoint(),
@@ -1918,7 +1918,11 @@ func (weh *workflowExecutionEventHandlerImpl) handleNexusOperationStarted(event
 	command := weh.commandsHelper.handleNexusOperationStarted(attributes.ScheduledEventId)
 	state := command.getData().(*scheduledNexusOperation)
 	if state.startedCallback != nil {
-		state.startedCallback(attributes.OperationId, nil)
+		token := attributes.OperationToken
+		if token == "" {
+			token = attributes.OperationId //lint:ignore SA1019 this field is sent by servers older than 1.27.0.
+		}
+		state.startedCallback(token, nil)
 		state.startedCallback = nil
 	}
 	return nil

internal/internal_nexus_task_handler.go

@@ -221,7 +221,7 @@ func (h *nexusTaskHandler) handleStartOperation(
 		if !panic {
 			nctx.Log.Error("Handler returned error while processing Nexus task", tagError, err)
 		}
-		var unsuccessfulOperationErr *nexus.UnsuccessfulOperationError
+		var unsuccessfulOperationErr *nexus.OperationError
 		err = convertKnownErrors(err)
 		if errors.As(err, &unsuccessfulOperationErr) {
 			failure, err := h.errorToFailure(unsuccessfulOperationErr.Cause)
@@ -260,13 +260,19 @@ func (h *nexusTaskHandler) handleStartOperation(
 				Type: nexusLink.Type,
 			})
 		}
+		token := t.OperationToken
+		//lint:ignore SA1019 this field might be set by users of older SDKs.
+		if t.OperationID != "" {
+			token = t.OperationID //lint:ignore SA1019 this field might be set by users of older SDKs.
+		}
 		return &nexuspb.Response{
 			Variant: &nexuspb.Response_StartOperation{
 				StartOperation: &nexuspb.StartOperationResponse{
 					Variant: &nexuspb.StartOperationResponse_AsyncSuccess{
 						AsyncSuccess: &nexuspb.StartOperationResponse_Async{
-							OperationId: t.OperationID,
-							Links:       links,
+							OperationToken: token,
+							OperationId:    token,
+							Links:          links,
 						},
 					},
 				},
@@ -313,7 +319,12 @@ func (h *nexusTaskHandler) handleCancelOperation(ctx context.Context, nctx *Nexu
 				nctx.Log.Error("Panic captured while handling Nexus task", tagStackTrace, string(debug.Stack()), tagError, err)
 			}
 		}()
-		err = h.nexusHandler.CancelOperation(ctx, req.GetService(), req.GetOperation(), req.GetOperationId(), cancelOptions)
+		token := req.GetOperationToken()
+		if token == "" {
+			// Support servers older than 1.27.0.
+			token = req.GetOperationId()
+		}
+		err = h.nexusHandler.CancelOperation(ctx, req.GetService(), req.GetOperation(), token, cancelOptions)
 	}()
 	if ctx.Err() != nil {
 		if !panic {
@@ -469,16 +480,13 @@ var emptyReaderNopCloser = io.NopCloser(bytes.NewReader([]byte{}))
 
 // convertKnownErrors converts known errors to corresponding Nexus HandlerError.
 func convertKnownErrors(err error) error {
-	// Handle common errors returned from various client methods.
-	if workflowErr, ok := err.(*WorkflowExecutionError); ok {
-		return nexus.NewFailedOperationError(workflowErr)
-	}
-	if queryRejectedErr, ok := err.(*QueryRejectedError); ok {
-		return nexus.NewFailedOperationError(queryRejectedErr)
-	}
 	// Not using errors.As to be consistent ApplicationError checking with the rest of the SDK.
 	if appErr, ok := err.(*ApplicationError); ok && appErr.NonRetryable() {
-		return nexus.NewFailedOperationError(appErr)
+		return &nexus.HandlerError{
+			// TODO(bergundy): Change this to a non retryable internal error after the 1.27.0 server release.
+			Type:  nexus.HandlerErrorTypeBadRequest,
+			Cause: appErr,
+		}
 	}
 	return convertServiceError(err)
 }
@@ -502,7 +510,10 @@ func convertServiceError(err error) error {
 	st = stGetter.Status()
 
 	switch st.Code() {
-	case codes.AlreadyExists, codes.InvalidArgument, codes.FailedPrecondition, codes.OutOfRange:
+	case codes.InvalidArgument:
+		return &nexus.HandlerError{Type: nexus.HandlerErrorTypeBadRequest, Cause: err}
+	case codes.AlreadyExists, codes.FailedPrecondition, codes.OutOfRange:
+		// TODO(bergundy): Change this to a non retryable internal error after the 1.27.0 server release.
 		return &nexus.HandlerError{Type: nexus.HandlerErrorTypeBadRequest, Cause: err}
 	case codes.Aborted, codes.Unavailable:
 		return &nexus.HandlerError{Type: nexus.HandlerErrorTypeUnavailable, Cause: err}

internal/internal_worker_base.go

@@ -99,7 +99,7 @@ type (
 		RequestCancelChildWorkflow(namespace, workflowID string)
 		RequestCancelExternalWorkflow(namespace, workflowID, runID string, callback ResultHandler)
 		ExecuteChildWorkflow(params ExecuteWorkflowParams, callback ResultHandler, startedHandler func(r WorkflowExecution, e error))
-		ExecuteNexusOperation(params executeNexusOperationParams, callback func(*commonpb.Payload, error), startedHandler func(opID string, e error)) int64
+		ExecuteNexusOperation(params executeNexusOperationParams, callback func(*commonpb.Payload, error), startedHandler func(token string, e error)) int64
 		RequestCancelNexusOperation(seq int64)
 		GetLogger() log.Logger
 		GetMetricsHandler() metrics.Handler

internal/internal_workflow_testsuite.go

@@ -106,7 +106,7 @@ type (
 		env             *testWorkflowEnvironmentImpl
 		seq             int64
 		params          executeNexusOperationParams
-		operationID     string
+		operationToken  string
 		cancelRequested bool
 		started         bool
 		done            bool
@@ -2504,7 +2504,7 @@ func (env *testWorkflowEnvironmentImpl) ExecuteNexusOperation(
 			}, true)
 		case *nexuspb.StartOperationResponse_AsyncSuccess:
 			env.postCallback(func() {
-				opID = v.AsyncSuccess.GetOperationId()
+				opID = v.AsyncSuccess.GetOperationToken()
 				handle.startedCallback(opID, nil)
 				if handle.cancelRequested {
 					handle.cancel()
@@ -2551,7 +2551,7 @@ func (env *testWorkflowEnvironmentImpl) RequestCancelNexusOperation(seq int64) {
 func (env *testWorkflowEnvironmentImpl) RegisterNexusAsyncOperationCompletion(
 	service string,
 	operation string,
-	operationID string,
+	token string,
 	result any,
 	err error,
 	delay time.Duration,
@@ -2586,7 +2586,7 @@ func (env *testWorkflowEnvironmentImpl) RegisterNexusAsyncOperationCompletion(
 	env.setNexusAsyncOperationCompletionHandle(
 		service,
 		operation,
-		operationID,
+		token,
 		&testNexusAsyncOperationHandle{
 			result: data,
 			err:    err,
@@ -2599,28 +2599,28 @@ func (env *testWorkflowEnvironmentImpl) RegisterNexusAsyncOperationCompletion(
 func (env *testWorkflowEnvironmentImpl) getNexusAsyncOperationCompletionHandle(
 	service string,
 	operation string,
-	operationID string,
+	token string,
 ) *testNexusAsyncOperationHandle {
-	uniqueOpID := env.makeUniqueNexusOperationID(service, operation, operationID)
+	uniqueOpID := env.makeUniqueNexusOperationToken(service, operation, token)
 	return env.nexusAsyncOpHandle[uniqueOpID]
 }
 
 func (env *testWorkflowEnvironmentImpl) setNexusAsyncOperationCompletionHandle(
 	service string,
 	operation string,
-	operationID string,
+	token string,
 	handle *testNexusAsyncOperationHandle,
 ) {
-	uniqueOpID := env.makeUniqueNexusOperationID(service, operation, operationID)
+	uniqueOpID := env.makeUniqueNexusOperationToken(service, operation, token)
 	env.nexusAsyncOpHandle[uniqueOpID] = handle
 }
 
 func (env *testWorkflowEnvironmentImpl) deleteNexusAsyncOperationCompletionHandle(
 	service string,
 	operation string,
-	operationID string,
+	token string,
 ) {
-	uniqueOpID := env.makeUniqueNexusOperationID(service, operation, operationID)
+	uniqueOpID := env.makeUniqueNexusOperationToken(service, operation, token)
 	delete(env.nexusAsyncOpHandle, uniqueOpID)
 }
 
@@ -2630,21 +2630,21 @@ func (env *testWorkflowEnvironmentImpl) scheduleNexusAsyncOperationCompletion(
 	completionHandle := env.getNexusAsyncOperationCompletionHandle(
 		handle.params.client.Service(),
 		handle.params.operation,
-		handle.operationID,
+		handle.operationToken,
 	)
 	if completionHandle == nil {
 		return
 	}
 	env.deleteNexusAsyncOperationCompletionHandle(
 		handle.params.client.Service(),
 		handle.params.operation,
-		handle.operationID,
+		handle.operationToken,
 	)
 	var nexusErr error
 	if completionHandle.err != nil {
 		nexusErr = env.failureConverter.FailureToError(nexusOperationFailure(
 			handle.params,
-			handle.operationID,
+			handle.operationToken,
 			&failurepb.Failure{
 				Message: completionHandle.err.Error(),
 				FailureInfo: &failurepb.Failure_ApplicationFailureInfo{
@@ -2670,7 +2670,7 @@ func (env *testWorkflowEnvironmentImpl) resolveNexusOperation(seq int64, result
 		}
 		if err != nil {
 			failure := env.failureConverter.ErrorToFailure(err)
-			err = env.failureConverter.FailureToError(nexusOperationFailure(handle.params, handle.operationID, failure.GetCause()))
+			err = env.failureConverter.FailureToError(nexusOperationFailure(handle.params, handle.operationToken, failure.GetCause()))
 			handle.completedCallback(nil, err)
 		} else {
 			handle.completedCallback(result, nil)
@@ -2696,12 +2696,12 @@ func (env *testWorkflowEnvironmentImpl) deleteNexusOperationHandle(seqID int64)
 	delete(env.runningNexusOperations, seqID)
 }
 
-func (env *testWorkflowEnvironmentImpl) makeUniqueNexusOperationID(
+func (env *testWorkflowEnvironmentImpl) makeUniqueNexusOperationToken(
 	service string,
 	operation string,
-	operationID string,
+	token string,
 ) string {
-	return fmt.Sprintf("%s_%s_%s", service, operation, operationID)
+	return fmt.Sprintf("%s_%s_%s", service, operation, token)
 }
 
 func (env *testWorkflowEnvironmentImpl) SideEffect(f func() (*commonpb.Payloads, error), callback ResultHandler) {
@@ -3202,9 +3202,9 @@ func (h *testNexusOperationHandle) newCancelTask() *workflowservice.PollNexusTas
 			Header:        h.params.nexusHeader,
 			Variant: &nexuspb.Request_CancelOperation{
 				CancelOperation: &nexuspb.CancelOperationRequest{
-					Service:     h.params.client.Service(),
-					Operation:   h.params.operation,
-					OperationId: h.operationID,
+					Service:        h.params.client.Service(),
+					Operation:      h.params.operation,
+					OperationToken: h.operationToken,
 				},
 			},
 		},
@@ -3241,7 +3241,7 @@ func (h *testNexusOperationHandle) startedCallback(opID string, e error) {
 		// Ignore duplciate starts.
 		return
 	}
-	h.operationID = opID
+	h.operationToken = opID
 	h.started = true
 	h.onStarted(opID, e)
 	h.env.runningCount--
@@ -3251,7 +3251,7 @@ func (h *testNexusOperationHandle) cancel() {
 	if h.done {
 		return
 	}
-	if h.started && h.operationID == "" {
+	if h.started && h.operationToken == "" {
 		panic(fmt.Errorf("incomplete operation has no operation ID: (%s, %s, %s)",
 			h.params.client.Endpoint(), h.params.client.Service(), h.params.operation))
 	}
@@ -3445,34 +3445,34 @@ func (r *testNexusHandler) CancelOperation(
 	ctx context.Context,
 	service string,
 	operation string,
-	operationID string,
+	token string,
 	options nexus.CancelOperationOptions,
 ) error {
 	if r.opHandle.isMocked {
 		// if the operation was mocked, then there's no workflow running
 		return nil
 	}
-	return r.handler.CancelOperation(ctx, service, operation, operationID, options)
+	return r.handler.CancelOperation(ctx, service, operation, token, options)
 }
 
 func (r *testNexusHandler) GetOperationInfo(
 	ctx context.Context,
 	service string,
 	operation string,
-	operationID string,
+	token string,
 	options nexus.GetOperationInfoOptions,
 ) (*nexus.OperationInfo, error) {
-	return r.handler.GetOperationInfo(ctx, service, operation, operationID, options)
+	return r.handler.GetOperationInfo(ctx, service, operation, token, options)
 }
 
 func (r *testNexusHandler) GetOperationResult(
 	ctx context.Context,
 	service string,
 	operation string,
-	operationID string,
+	token string,
 	options nexus.GetOperationResultOptions,
 ) (any, error) {
-	return r.handler.GetOperationResult(ctx, service, operation, operationID, options)
+	return r.handler.GetOperationResult(ctx, service, operation, token, options)
 }
 
 func (env *testWorkflowEnvironmentImpl) registerNexusOperationReference(