Execute nexus operation from a workflow (#1473)

* Execute nexus operation from a workflow

* Address review comments

* Add test environment support for Nexus Operations (#1475)

* Add test environment support for Nexus Operations

* Change client to not allow any direct calls

interceptor/interceptor.go

@@ -131,6 +131,11 @@ type HandleQueryInput = internal.HandleQueryInput
 // NOTE: Experimental
 type UpdateInput = internal.UpdateInput
 
+// RequestCancelNexusOperationInput is the input to WorkflowOutboundInterceptor.RequestCancelNexusOperation.
+//
+// NOTE: Experimental
+type RequestCancelNexusOperationInput = internal.RequestCancelNexusOperationInput
+
 // WorkflowOutboundInterceptor is an interface for all workflow calls
 // originating from the SDK.
 //

internal/client.go

@@ -31,7 +31,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"go.temporal.io/api/common/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/operatorservice/v1"
@@ -649,7 +648,7 @@ type (
 		// request ID. Only settable by the SDK - e.g. [temporalnexus.workflowRunOperation].
 		requestID string
 		// workflow completion callback. Only settable by the SDK - e.g. [temporalnexus.workflowRunOperation].
-		callbacks []*common.Callback
+		callbacks []*commonpb.Callback
 	}
 
 	// RetryPolicy defines the retry policy.
@@ -1004,6 +1003,6 @@ func SetRequestIDOnStartWorkflowOptions(opts *StartWorkflowOptions, requestID st
 
 // SetCallbacksOnStartWorkflowOptions is an internal only method for setting callbacks on StartWorkflowOptions.
 // Callbacks are purposefully not exposed to users for the time being.
-func SetCallbacksOnStartWorkflowOptions(opts *StartWorkflowOptions, callbacks []*common.Callback) {
+func SetCallbacksOnStartWorkflowOptions(opts *StartWorkflowOptions, callbacks []*commonpb.Callback) {
 	opts.callbacks = callbacks
 }

internal/error.go

@@ -255,6 +255,28 @@ type (
 		cause            error
 	}
 
+	// NexusOperationError is an error returned when a Nexus Operation has failed.
+	//
+	// NOTE: Experimental
+	NexusOperationError struct {
+		// The raw proto failure object this error was created from.
+		Failure *failurepb.Failure
+		// Error message.
+		Message string
+		// ID of the NexusOperationScheduled event.
+		ScheduledEventID int64
+		// Endpoint name.
+		Endpoint string
+		// Service name.
+		Service string
+		// Operation name.
+		Operation string
+		// Operation ID - may be empty if the operation completed synchronously.
+		OperationID string
+		// Chained cause - typically an ApplicationError or a CanceledError.
+		Cause error
+	}
+
 	// ChildWorkflowExecutionAlreadyStartedError is set as the cause of
 	// ChildWorkflowExecutionError when failure is due the child workflow having
 	// already started.
@@ -800,6 +822,32 @@ func (e *ChildWorkflowExecutionError) Unwrap() error {
 	return e.cause
 }
 
+// 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)
+	if e.Cause != nil {
+		msg = fmt.Sprintf("%s: %v", msg, e.Cause)
+	}
+	return msg
+}
+
+// setFailure implements the failureHolder interface for consistency with other failure based errors..
+func (e *NexusOperationError) setFailure(f *failurepb.Failure) {
+	e.Failure = f
+}
+
+// failure implements the failureHolder interface for consistency with other failure based errors.
+func (e *NexusOperationError) failure() *failurepb.Failure {
+	return e.Failure
+}
+
+// Unwrap returns the Cause associated with this error.
+func (e *NexusOperationError) Unwrap() error {
+	return e.Cause
+}
+
 // Error from error interface
 func (*NamespaceNotFoundError) Error() string {
 	return "namespace not found"

internal/failure_converter.go

@@ -160,6 +160,15 @@ func (dfc *DefaultFailureConverter) ErrorToFailure(err error) *failurepb.Failure
 			RetryState:       err.retryState,
 		}
 		failure.FailureInfo = &failurepb.Failure_ChildWorkflowExecutionFailureInfo{ChildWorkflowExecutionFailureInfo: failureInfo}
+	case *NexusOperationError:
+		failureInfo := &failurepb.NexusOperationFailureInfo{
+			ScheduledEventId: err.ScheduledEventID,
+			Endpoint: err.Endpoint,
+			Service: err.Service,
+			Operation: err.Operation,
+			OperationId: err.OperationID,
+		}
+		failure.FailureInfo = &failurepb.Failure_NexusOperationExecutionFailureInfo{NexusOperationExecutionFailureInfo: failureInfo}
 	default: // All unknown errors are considered to be retryable ApplicationFailureInfo.
 		failureInfo := &failurepb.ApplicationFailureInfo{
 			Type:         getErrType(err),
@@ -254,6 +263,17 @@ func (dfc *DefaultFailureConverter) FailureToError(failure *failurepb.Failure) e
 			childWorkflowExecutionFailureInfo.GetRetryState(),
 			dfc.FailureToError(failure.GetCause()),
 		)
+	} else if info := failure.GetNexusOperationExecutionFailureInfo(); info != nil {
+		err = &NexusOperationError{
+			Message:          failure.Message,
+			Cause:            dfc.FailureToError(failure.GetCause()),
+			Failure:          originalFailure,
+			ScheduledEventID: info.GetScheduledEventId(),
+			Endpoint:         info.GetEndpoint(),
+			Service:          info.GetService(),
+			Operation:        info.GetOperation(),
+			OperationID:      info.GetOperationId(),
+		}
 	}
 
 	if err == nil {

internal/interceptor.go

@@ -169,6 +169,20 @@ type HandleQueryInput struct {
 	Args      []interface{}
 }
 
+// RequestCancelNexusOperationInput is the input to WorkflowOutboundInterceptor.RequestCancelNexusOperation.
+//
+// NOTE: Experimental
+type RequestCancelNexusOperationInput struct {
+	// Client that was used to start the operation.
+	Client NexusClient
+	// Operation name.
+	Operation any
+	// Operation ID. May be empty if the operation is synchronous or has not started yet.
+	ID string
+	// seq number. For internal use only.
+	seq int64
+}
+
 // WorkflowOutboundInterceptor is an interface for all workflow calls
 // originating from the SDK. See documentation in the interceptor package for
 // more details.
@@ -283,6 +297,15 @@ type WorkflowOutboundInterceptor interface {
 	// interceptor.WorkflowHeader will return a non-nil map for this context.
 	NewContinueAsNewError(ctx Context, wfn interface{}, args ...interface{}) error
 
+	// ExecuteNexusOperation intercepts NexusClient.ExecuteOperation.
+	//
+	// NOTE: Experimental
+	ExecuteNexusOperation(ctx Context, client NexusClient, operation any, input any, options NexusOperationOptions) NexusOperationFuture
+	// RequestCancelNexusOperation intercepts Nexus Operation cancelation via context.
+	//
+	// NOTE: Experimental
+	RequestCancelNexusOperation(ctx Context, input RequestCancelNexusOperationInput)
+
 	mustEmbedWorkflowOutboundInterceptorBase()
 }
 

internal/interceptor_base.go

@@ -389,6 +389,24 @@ func (w *WorkflowOutboundInterceptorBase) NewContinueAsNewError(
 	return w.Next.NewContinueAsNewError(ctx, wfn, args...)
 }
 
+// ExecuteNexusOperation implements
+// WorkflowOutboundInterceptor.ExecuteNexusOperation.
+func (w *WorkflowOutboundInterceptorBase) ExecuteNexusOperation(
+	ctx Context,
+	client NexusClient,
+	operation any,
+	input any,
+	options NexusOperationOptions,
+) NexusOperationFuture {
+	return w.Next.ExecuteNexusOperation(ctx, client, operation, input, options)
+}
+
+// RequestCancelNexusOperation implements
+// WorkflowOutboundInterceptor.RequestCancelNexusOperation.
+func (w *WorkflowOutboundInterceptorBase) RequestCancelNexusOperation(ctx Context, input RequestCancelNexusOperationInput) {
+	w.Next.RequestCancelNexusOperation(ctx, input)
+}
+
 func (*WorkflowOutboundInterceptorBase) mustEmbedWorkflowOutboundInterceptorBase() {}
 
 // ClientInterceptorBase is a default implementation of ClientInterceptor meant