Support conflict options for starting Nexus operations in test framework (#1828)

* Support conflict options for starting Nexus operations in test framework

* fix nexus error handling

* address comments

* simplify test

* address comments

* Update cli version

* fix test

Author: rodrigozhou

internal/cmd/build/main.go

@@ -142,7 +142,7 @@ func (b *builder) integrationTest() error {
 		devServer, err := testsuite.StartDevServer(context.Background(), testsuite.DevServerOptions{
 			// TODO: Use stable release once server 1.27.0 is out.
 			CachedDownload: testsuite.CachedDownload{
-				Version: "v1.3.0-versioning.0",
+				Version: "v1.3.0-rc.0",
 			},
 			ClientOptions: &client.Options{
 				HostPort:  "127.0.0.1:7233",

internal/internal_worker_base.go

@@ -232,6 +232,13 @@ type (
 	}
 )
 
+func (h ResultHandler) wrap(callback ResultHandler) ResultHandler {
+	return func(result *commonpb.Payloads, err error) {
+		callback(result, err)
+		h(result, err)
+	}
+}
+
 func (t *polledTask) getTask() taskForWorker {
 	return t.task
 }

internal/internal_workflow.go

@@ -214,7 +214,10 @@ type (
 		WorkflowID               string
 		WaitForCancellation      bool
 		WorkflowIDReusePolicy    enumspb.WorkflowIdReusePolicy
+		// WorkflowIDConflictPolicy and OnConflictOptions are only used in test environment for
+		// running Nexus operations as child workflow.
 		WorkflowIDConflictPolicy enumspb.WorkflowIdConflictPolicy
+		OnConflictOptions        *OnConflictOptions
 		DataConverter            converter.DataConverter
 		RetryPolicy              *commonpb.RetryPolicy
 		CronSchedule             string

internal/internal_workflow_testsuite.go

@@ -428,7 +428,11 @@ func (env *testWorkflowEnvironmentImpl) setContinuedExecutionRunID(rid string) {
 	env.workflowInfo.ContinuedExecutionRunID = rid
 }
 
-func (env *testWorkflowEnvironmentImpl) newTestWorkflowEnvironmentForChild(params *ExecuteWorkflowParams, callback ResultHandler, startedHandler func(r WorkflowExecution, e error)) (*testWorkflowEnvironmentImpl, error) {
+func (env *testWorkflowEnvironmentImpl) newTestWorkflowEnvironmentForChild(
+	params *ExecuteWorkflowParams,
+	callback ResultHandler,
+	startedHandler func(r WorkflowExecution, e error),
+) (*testWorkflowEnvironmentImpl, error) {
 	// create a new test env
 	childEnv := newTestWorkflowEnvironmentImpl(env.testSuite, env.registry)
 	childEnv.parentEnv = env
@@ -474,15 +478,27 @@ func (env *testWorkflowEnvironmentImpl) newTestWorkflowEnvironmentForChild(param
 
 	childEnv.runTimeout = params.WorkflowRunTimeout
 	if workflowHandler, ok := env.runningWorkflows[params.WorkflowID]; ok {
+		alreadyStartedErr := serviceerror.NewWorkflowExecutionAlreadyStarted(
+			"Workflow execution already started",
+			"",
+			childEnv.workflowInfo.WorkflowExecution.RunID,
+		)
 		// duplicate workflow ID
 		if !workflowHandler.handled {
-			return nil, serviceerror.NewWorkflowExecutionAlreadyStarted("Workflow execution already started", "", "")
+			if params.WorkflowIDConflictPolicy == enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING {
+				if params.OnConflictOptions != nil && params.OnConflictOptions.AttachCompletionCallbacks {
+					workflowHandler.callback = workflowHandler.callback.wrap(callback)
+				}
+				startedHandler(workflowHandler.env.workflowInfo.WorkflowExecution, nil)
+				return nil, nil
+			}
+			return nil, alreadyStartedErr
 		}
 		if params.WorkflowIDReusePolicy == enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE {
-			return nil, serviceerror.NewWorkflowExecutionAlreadyStarted("Workflow execution already started", "", "")
+			return nil, alreadyStartedErr
 		}
 		if workflowHandler.err == nil && params.WorkflowIDReusePolicy == enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY {
-			return nil, serviceerror.NewWorkflowExecutionAlreadyStarted("Workflow execution already started", "", "")
+			return nil, alreadyStartedErr
 		}
 	}
 
@@ -2380,16 +2396,20 @@ func (env *testWorkflowEnvironmentImpl) executeChildWorkflowWithDelay(delayStart
 	childEnv, err := env.newTestWorkflowEnvironmentForChild(&params, callback, startedHandler)
 	if err != nil {
 		env.logger.Info("ExecuteChildWorkflow failed", tagError, err)
-		callback(nil, err)
 		startedHandler(WorkflowExecution{}, err)
+		callback(nil, err)
 		return
 	}
 
-	env.logger.Info("ExecuteChildWorkflow", tagWorkflowType, params.WorkflowType.Name)
-	env.runningCount++
+	// childEnv can be nil when WorkflowIDConflictPolicy is USE_EXISTING and there's already a running
+	// workflow. This is only possible in the test environment for running Nexus handler workflow.
+	if childEnv != nil {
+		env.logger.Info("ExecuteChildWorkflow", tagWorkflowType, params.WorkflowType.Name)
+		env.runningCount++
 
-	// run child workflow in separate goroutinue
-	go childEnv.executeWorkflowInternal(delayStart, params.WorkflowType.Name, params.Input)
+		// run child workflow in separate goroutinue
+		go childEnv.executeWorkflowInternal(delayStart, params.WorkflowType.Name, params.Input)
+	}
 }
 
 func (env *testWorkflowEnvironmentImpl) newTestNexusTaskHandler(

internal/nexus_operations.go

@@ -453,6 +453,7 @@ func (t *testSuiteClientForNexusOperations) ExecuteWorkflow(ctx context.Context,
 	}
 
 	run := &testEnvWorkflowRunForNexusOperations{}
+	startedErrCh := make(chan error, 1)
 	doneCh := make(chan error)
 
 	var callback *commonpb.Callback
@@ -476,6 +477,8 @@ func (t *testSuiteClientForNexusOperations) ExecuteWorkflow(ctx context.Context,
 				WorkflowTaskTimeout:      options.WorkflowTaskTimeout,
 				DataConverter:            t.env.dataConverter,
 				WorkflowIDReusePolicy:    options.WorkflowIDReusePolicy,
+				WorkflowIDConflictPolicy: options.WorkflowIDConflictPolicy,
+				OnConflictOptions:        options.onConflictOptions,
 				ContextPropagators:       t.env.contextPropagators,
 				SearchAttributes:         options.SearchAttributes,
 				TypedSearchAttributes:    options.TypedSearchAttributes,
@@ -485,6 +488,14 @@ func (t *testSuiteClientForNexusOperations) ExecuteWorkflow(ctx context.Context,
 				RetryPolicy:              convertToPBRetryPolicy(options.RetryPolicy),
 			},
 		}, func(result *commonpb.Payloads, wfErr error) {
+			// This callback handles async completion of Nexus operations. If there was an error when
+			// starting the workflow, then the operation failed synchronously and this callback doesn't
+			// need to be executed.
+			startedErr := <-startedErrCh
+			if startedErr != nil {
+				return
+			}
+
 			ncb := callback.GetNexus()
 			if ncb == nil {
 				return
@@ -519,6 +530,8 @@ func (t *testSuiteClientForNexusOperations) ExecuteWorkflow(ctx context.Context,
 			}
 		}, func(r WorkflowExecution, err error) {
 			run.WorkflowExecution = r
+			startedErrCh <- err
+			close(startedErrCh)
 			doneCh <- err
 		})
 	}, false)