Add non-deterministic workflow option (#475)

This commit adds an option to configure how worker execution should
handle detection of a non-deterministic workflow.

Currently, when non-determinism is detected in execution history, the
task handler on the client simply logs the error to console *without*
replying back to the cadence server, which leads the server to think
the decision task has timed out and thus reschedules a new decsion task.

The currently behavior is less than satisfactory for 2 reasons:
1. From server side there's no indication of the non-determinism error.
   It simply looks like a timeout.
2. From the client side, there's no recover option in that the client
   will just pull off the new rescheduled decision task and fail again
   and again.

This commit adds a new option to cadence workers to explicit fail the
workflow where a mismatched history has been detected. This should
yield a more robust debugging experience for customers whose workflows
fail due to non-determinism.

Author: arthurgan

internal/internal_task_handlers.go

@@ -98,15 +98,16 @@ type (
 
 	// workflowTaskHandlerImpl is the implementation of WorkflowTaskHandler
 	workflowTaskHandlerImpl struct {
-		domain                 string
-		metricsScope           tally.Scope
-		ppMgr                  pressurePointMgr
-		logger                 *zap.Logger
-		identity               string
-		enableLoggingInReplay  bool
-		disableStickyExecution bool
-		hostEnv                *hostEnvImpl
-		laTunnel               *localActivityTunnel
+		domain                         string
+		metricsScope                   tally.Scope
+		ppMgr                          pressurePointMgr
+		logger                         *zap.Logger
+		identity                       string
+		enableLoggingInReplay          bool
+		disableStickyExecution         bool
+		hostEnv                        *hostEnvImpl
+		laTunnel                       *localActivityTunnel
+		nonDeterministicWorkflowPolicy NonDeterministicWorkflowPolicy
 	}
 
 	activityProvider func(name string) activity
@@ -307,6 +308,7 @@ func newWorkflowTaskHandler(
 		enableLoggingInReplay:  params.EnableLoggingInReplay,
 		disableStickyExecution: params.DisableStickyExecution,
 		hostEnv:                hostEnv,
+		nonDeterministicWorkflowPolicy: params.NonDeterministicWorkflowPolicy,
 	}
 }
 
@@ -653,7 +655,26 @@ ProcessEvents:
 		// check if decisions from reply matches to the history events
 		if err := matchReplayWithHistory(replayDecisions, respondEvents); err != nil {
 			wth.logger.Error("Replay and history mismatch.", zap.Error(err))
-			return nil, "", err
+
+			// Whether or not we store the error in workflowContext.err makes
+			// a significant difference, to the point that it affects client's observable
+			// behavior as far as handling non-deterministic workflows.
+			//
+			// If we store it in workflowContext.err, the decision task completion code
+			// will pick up the error and correctly wrap it in the response request we sent back
+			// to the server, which in this case will contain a request to fail the workflow.
+			//
+			// If we simply return the error, the decision task completion code path will not
+			// execute at all, therefore, no response is sent back to the server and we will
+			// look like a decision task time out.
+			switch wth.nonDeterministicWorkflowPolicy {
+			case NonDeterministicWorkflowPolicyFailWorkflow:
+				eventHandler.Complete(nil, NewCustomError("nondeterministic workflow", err.Error()))
+			case NonDeterministicWorkflowPolicyBlockWorkflow:
+				return nil, "", err
+			default:
+				panic(fmt.Sprintf("unknown mismatched workflow history policy."))
+			}
 		}
 	}
 

internal/internal_task_handlers_test.go

@@ -364,6 +364,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 		TaskList: taskList,
 		Identity: "test-id-1",
 		Logger:   zap.NewNop(),
+		NonDeterministicWorkflowPolicy: NonDeterministicWorkflowPolicyBlockWorkflow,
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
 	request, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
@@ -380,6 +381,26 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 	t.Nil(request)
 	t.Contains(err.Error(), "nondeterministic")
 
+	// now, create a new task handler with fail nondeterministic workflow policy
+	// and verify that it handles the mismatching history correctly.
+	params.NonDeterministicWorkflowPolicy = NonDeterministicWorkflowPolicyFailWorkflow
+	failOnNondeterminismTaskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
+	task = createWorkflowTask(testEvents, 3, "HelloWorld_Workflow")
+	request, _, err = failOnNondeterminismTaskHandler.ProcessWorkflowTask(task, nil, false)
+	// When FailWorkflow policy is set, task handler does not return an error,
+	// because it will indicate non determinism in the request.
+	t.NoError(err)
+	// Verify that request is a RespondDecisionTaskCompleteRequest
+	response, ok := request.(*s.RespondDecisionTaskCompletedRequest)
+	t.True(ok)
+	// Verify there's at least 1 decision
+	// and the last last decision is to fail workflow
+	// and contains proper justification.(i.e. nondeterminism).
+	t.True(len(response.Decisions) > 0)
+	closeDecision := response.Decisions[len(response.Decisions)-1]
+	t.Equal(*closeDecision.DecisionType, s.DecisionTypeFailWorkflowExecution)
+	t.Contains(*closeDecision.FailWorkflowExecutionDecisionAttributes.Reason, "nondeterministic")
+
 	// now with different package name to activity type
 	testEvents[4].ActivityTaskScheduledEventAttributes.ActivityType.Name = common.StringPtr("new-package.Greeter_Activity")
 	task = createWorkflowTask(testEvents, 3, "HelloWorld_Workflow")

internal/internal_worker.go

@@ -141,6 +141,10 @@ type (
 		DisableStickyExecution bool
 
 		StickyScheduleToStartTimeout time.Duration
+
+		// NonDeterministicWorkflowPolicy is used for configuring how client's decision task handler deals with
+		// mismatched history events (presumably arising from non-deterministic workflow definitions).
+		NonDeterministicWorkflowPolicy NonDeterministicWorkflowPolicy
 	}
 )
 
@@ -1017,6 +1021,7 @@ func newAggregatedWorker(
 		DisableStickyExecution:               wOptions.DisableStickyExecution,
 		StickyScheduleToStartTimeout:         wOptions.StickyScheduleToStartTimeout,
 		TaskListActivitiesPerSecond:          wOptions.TaskListActivitiesPerSecond,
+		NonDeterministicWorkflowPolicy:       wOptions.NonDeterministicWorkflowPolicy,
 	}
 
 	ensureRequiredParams(&workerParams)

internal/worker.go

@@ -137,9 +137,32 @@ type (
 		// Optional: sets context for activity. The context can be used to pass any configuration to activity
 		// like common logger for all activities.
 		BackgroundActivityContext context.Context
+
+		// NonDeterministicWorkflowPolicy is used for configuring how decision worker deals with non-deterministic history events
+		// (presumably arising from non-deterministic workflow definitions or non-backward compatible workflow definition changes).
+		NonDeterministicWorkflowPolicy NonDeterministicWorkflowPolicy
 	}
 )
 
+// NonDeterministicWorkflowPolicy is an enum for configuring how client's decision task handler deals with
+// mismatched history events (presumably arising from non-deterministic workflow definitions).
+type NonDeterministicWorkflowPolicy int
+
+const (
+	// NonDeterministicWorkflowPolicyBlockWorkflow is the default policy for handling detected non-determinism.
+	// This option simply logs to console with an error message that non-determinism is detected, but
+	// does *NOT* reply anything back to the server.
+	// It is chosen as default for backward compatibility reasons because it preserves the old behavior
+	// for handling non-determinism that we had before NonDeterministicWorkflowPolicy type was added to
+	// allow more configurability.
+	NonDeterministicWorkflowPolicyBlockWorkflow NonDeterministicWorkflowPolicy = iota
+	// NonDeterministicWorkflowPolicyFailWorkflow behaves exactly the same as Ignore, up until the very
+	// end of processing a decision task.
+	// Whereas default does *NOT* reply anything back to the server, fail workflow replies back with a request
+	// to fail the workflow execution.
+	NonDeterministicWorkflowPolicyFailWorkflow
+)
+
 // NewWorker creates an instance of worker for managing workflow and activity executions.
 // service 	- thrift connection to the cadence server.
 // domain - the name of the cadence domain.

worker/worker.go

@@ -37,6 +37,25 @@ type (
 
 	// Options is used to configure a worker instance.
 	Options = internal.WorkerOptions
+
+	// NonDeterministicWorkflowPolicy is an enum for configuring how client's decision task handler deals with
+	// mismatched history events (presumably arising from non-deterministic workflow definitions).
+	NonDeterministicWorkflowPolicy = internal.NonDeterministicWorkflowPolicy
+)
+
+const (
+	// NonDeterministicWorkflowPolicyBlockWorkflow is the default policy for handling detected non-determinism.
+	// This option simply logs to console with an error message that non-determinism is detected, but
+	// does *NOT* reply anything back to the server.
+	// It is chosen as default for backward compatibility reasons because it preserves the old behavior
+	// for handling non-determinism that we had before NonDeterministicWorkflowPolicy type was added to
+	// allow more configurability.
+	NonDeterministicWorkflowPolicyBlockWorkflow = internal.NonDeterministicWorkflowPolicyBlockWorkflow
+	// NonDeterministicWorkflowPolicyFailWorkflow behaves exactly the same as Ignore, up until the very
+	// end of processing a decision task.
+	// Whereas default does *NOT* reply anything back to the server, fail workflow replies back with a request
+	// to fail the workflow execution.
+	NonDeterministicWorkflowPolicyFailWorkflow = internal.NonDeterministicWorkflowPolicyFailWorkflow
 )
 
 // New creates an instance of worker for managing workflow and activity executions.