Honor non-determinism fail workflow policy (#1287)

What changed?
Users can specify NonDeterministicWorkflowPolicy in worker options. If the FailWorkflow policy is chosen the workflow is expected to terminate as soon as it ends up with a nondeterministic state (e.g. activity order changed).
However this wasn't honored for a category of nondeterminism cases. This PR addresses it and workflows fail once any nondeterminism scenario is encountered.

There are two categories of nondeterminism cases in terms of how they get detected by client library:
1. Issue bubbles up as illegal state panic to the task handler. Most actual prod cases.
2. Issue is caught when comparing replay decisions with history. Replay test scenarios and a subset of prod cases.

FailWorkflow policy was honored for 2 but not for 1.

Why?
To make NonDeterministicWorkflowPolicy feature correct/complete.

How did you test it?
Added an integration test to simulate this scenario.

Author: taylanisikdemir

internal/common/metrics/scope.go

@@ -195,7 +195,7 @@ func (s *replayAwareScope) Capabilities() tally.Capabilities {
 }
 
 // GetTaggedScope return a scope with one or multiple tags,
-// input should be key value pairs like: GetTaggedScope(scope, tag1, val1, tag2, val2).
+// input should be key value pairs like: GetTaggedScope(tag1, val1, tag2, val2).
 func (ts *TaggedScope) GetTaggedScope(keyValueinPairs ...string) tally.Scope {
 	if len(keyValueinPairs)%2 != 0 {
 		panic("GetTaggedScope key value are not in pairs")

internal/common/metrics/service_wrapper.go

@@ -116,7 +116,7 @@ func (w *workflowServiceMetricsWrapper) getOperationScope(scopeName string) *ope
 }
 
 func (s *operationScope) handleError(err error) {
-	s.scope.Timer(CadenceLatency).Record(time.Now().Sub(s.startTime))
+	s.scope.Timer(CadenceLatency).Record(time.Since(s.startTime))
 	if err != nil {
 		switch err.(type) {
 		case *shared.EntityNotExistsError,

internal/internal_logging_tags.go

@@ -21,26 +21,34 @@
 package internal
 
 const (
-	tagActivityID            = "ActivityID"
-	tagActivityType          = "ActivityType"
-	tagDomain                = "Domain"
-	tagEventID               = "EventID"
-	tagEventType             = "EventType"
-	tagRunID                 = "RunID"
-	tagTaskList              = "TaskList"
-	tagTimerID               = "TimerID"
-	tagWorkflowID            = "WorkflowID"
-	tagWorkflowType          = "WorkflowType"
-	tagWorkerID              = "WorkerID"
-	tagWorkerType            = "WorkerType"
-	tagSideEffectID          = "SideEffectID"
-	tagChildWorkflowID       = "ChildWorkflowID"
-	tagLocalActivityType     = "LocalActivityType"
-	tagLocalActivityID       = "LocalActivityID"
-	tagQueryType             = "QueryType"
-	tagVisibilityQuery       = "VisibilityQuery"
-	tagPanicError            = "PanicError"
-	tagPanicStack            = "PanicStack"
-	causeTag                 = "pollerrorcause"
-	tagworkflowruntimelength = "workflowruntimelength"
+	tagActivityID                  = "ActivityID"
+	tagActivityType                = "ActivityType"
+	tagDomain                      = "Domain"
+	tagEventID                     = "EventID"
+	tagEventType                   = "EventType"
+	tagRunID                       = "RunID"
+	tagTaskList                    = "TaskList"
+	tagTimerID                     = "TimerID"
+	tagWorkflowID                  = "WorkflowID"
+	tagWorkflowType                = "WorkflowType"
+	tagWorkerID                    = "WorkerID"
+	tagWorkerType                  = "WorkerType"
+	tagSideEffectID                = "SideEffectID"
+	tagChildWorkflowID             = "ChildWorkflowID"
+	tagLocalActivityType           = "LocalActivityType"
+	tagLocalActivityID             = "LocalActivityID"
+	tagQueryType                   = "QueryType"
+	tagVisibilityQuery             = "VisibilityQuery"
+	tagPanicError                  = "PanicError"
+	tagPanicStack                  = "PanicStack"
+	causeTag                       = "pollerrorcause"
+	tagWorkflowRuntimeLength       = "workflowruntimelength"
+	tagNonDeterminismDetectionType = "NonDeterminismDetectionType"
+)
+
+type nonDeterminismDetectionType string
+
+const (
+	nonDeterminismDetectionTypeIllegalStatePanic nonDeterminismDetectionType = "illegalstatepanic"
+	nonDeterminismDetectionTypeReplayComparison  nonDeterminismDetectionType = "replaycomparison"
 )

internal/internal_task_handlers.go

@@ -35,6 +35,7 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
@@ -674,12 +675,12 @@ func (wth *workflowTaskHandlerImpl) getOrCreateWorkflowContext(
 	historyIterator HistoryIterator,
 ) (workflowContext *workflowExecutionContextImpl, err error) {
 	metricsScope := wth.metricsScope.GetTaggedScope(tagWorkflowType, task.WorkflowType.GetName())
-	defer func() {
+	defer func(metricsScope tally.Scope) {
 		if err == nil && workflowContext != nil && workflowContext.laTunnel == nil {
 			workflowContext.laTunnel = wth.laTunnel
 		}
 		metricsScope.Gauge(metrics.StickyCacheSize).Update(float64(getWorkflowCache().Size()))
-	}()
+	}(metricsScope)
 
 	runID := task.WorkflowExecution.GetRunId()
 
@@ -694,14 +695,13 @@ func (wth *workflowTaskHandlerImpl) getOrCreateWorkflowContext(
 	if workflowContext != nil {
 		workflowContext.Lock()
 		// add new tag on metrics scope with workflow runtime length category
-		executionRuntimeType := workflowCategorizedByTimeout(workflowContext.workflowInfo.ExecutionStartToCloseTimeoutSeconds)
-		metricsScope = metricsScope.Tagged(map[string]string{tagworkflowruntimelength: executionRuntimeType})
+		scope := metricsScope.Tagged(map[string]string{tagWorkflowRuntimeLength: workflowCategorizedByTimeout(workflowContext)})
 		if task.Query != nil && !isFullHistory {
 			// query task and we have a valid cached state
-			metricsScope.Counter(metrics.StickyCacheHit).Inc(1)
+			scope.Counter(metrics.StickyCacheHit).Inc(1)
 		} else if history.Events[0].GetEventId() == workflowContext.previousStartedEventID+1 {
 			// non query task and we have a valid cached state
-			metricsScope.Counter(metrics.StickyCacheHit).Inc(1)
+			scope.Counter(metrics.StickyCacheHit).Inc(1)
 		} else {
 			// non query task and cached state is missing events, we need to discard the cached state and rebuild one.
 			workflowContext.ResetIfStale(task, historyIterator)
@@ -949,39 +949,31 @@ ProcessEvents:
 	// the replay of that event will panic on the decision state machine and the workflow will be marked as completed
 	// with the panic error.
 	var nonDeterministicErr error
+	var nonDeterminismType nonDeterminismDetectionType
 	if !skipReplayCheck && !w.isWorkflowCompleted || isReplayTest {
 		// check if decisions from reply matches to the history events
 		if err := matchReplayWithHistory(w.workflowInfo, replayDecisions, respondEvents); err != nil {
 			nonDeterministicErr = err
+			nonDeterminismType = nonDeterminismDetectionTypeReplayComparison
 		}
-	}
-	if nonDeterministicErr == nil && w.err != nil {
-		if panicErr, ok := w.err.(*workflowPanicError); ok && panicErr.value != nil {
-			if _, isStateMachinePanic := panicErr.value.(stateMachineIllegalStatePanic); isStateMachinePanic {
-				if isReplayTest {
-					// NOTE: we should do this regardless if it's in replay test or not
-					// but since previously we checked the wrong error type, it may break existing customers workflow
-					// the issue is that we change the error type and that we change the error message, the customers
-					// are checking the error string - we plan to wrap all errors to avoid this issue in client v2
-					nonDeterministicErr = panicErr
-				} else {
-					// Since we know there is an error, we do the replay check to give more context in the log
-					replayErr := matchReplayWithHistory(w.workflowInfo, replayDecisions, respondEvents)
-					w.wth.logger.Error("Ignored workflow panic error",
-						zap.String(tagWorkflowType, task.WorkflowType.GetName()),
-						zap.String(tagWorkflowID, task.WorkflowExecution.GetWorkflowId()),
-						zap.String(tagRunID, task.WorkflowExecution.GetRunId()),
-						zap.Error(nonDeterministicErr),
-						zap.NamedError("ReplayError", replayErr),
-					)
-				}
-			}
-		}
+	} else if panicErr, ok := w.getWorkflowPanicIfIllegaleStatePanic(); ok {
+		// This is a nondeterministic execution which ended up panicking
+		nonDeterministicErr = panicErr
+		nonDeterminismType = nonDeterminismDetectionTypeIllegalStatePanic
+		// Since we know there is an error, we do the replay check to give more context in the log
+		replayErr := matchReplayWithHistory(w.workflowInfo, replayDecisions, respondEvents)
+		w.wth.logger.Error("Illegal state caused panic",
+			zap.String(tagWorkflowType, task.WorkflowType.GetName()),
+			zap.String(tagWorkflowID, task.WorkflowExecution.GetWorkflowId()),
+			zap.String(tagRunID, task.WorkflowExecution.GetRunId()),
+			zap.Error(nonDeterministicErr),
+			zap.NamedError("ReplayError", replayErr),
+		)
 	}
 
 	if nonDeterministicErr != nil {
-
-		w.wth.metricsScope.GetTaggedScope(tagWorkflowType, task.WorkflowType.GetName()).Counter(metrics.NonDeterministicError).Inc(1)
+		scope := w.wth.metricsScope.GetTaggedScope(tagWorkflowType, task.WorkflowType.GetName(), tagNonDeterminismDetectionType, string(nonDeterminismType))
+		scope.Counter(metrics.NonDeterministicError).Inc(1)
 		w.wth.logger.Error("non-deterministic-error",
 			zap.String(tagWorkflowType, task.WorkflowType.GetName()),
 			zap.String(tagWorkflowID, task.WorkflowExecution.GetWorkflowId()),
@@ -998,7 +990,7 @@ ProcessEvents:
 			// workflow timeout.
 			return nil, nonDeterministicErr
 		default:
-			panic(fmt.Sprintf("unknown mismatched workflow history policy."))
+			panic("unknown mismatched workflow history policy.")
 		}
 	}
 
@@ -1205,6 +1197,24 @@ func (w *workflowExecutionContextImpl) GetDecisionTimeout() time.Duration {
 	return time.Second * time.Duration(w.workflowInfo.TaskStartToCloseTimeoutSeconds)
 }
 
+func (w *workflowExecutionContextImpl) getWorkflowPanicIfIllegaleStatePanic() (*workflowPanicError, bool) {
+	if !w.isWorkflowCompleted || w.err == nil {
+		return nil, false
+	}
+
+	panicErr, ok := w.err.(*workflowPanicError)
+	if !ok || panicErr.value == nil {
+		return nil, false
+	}
+
+	_, ok = panicErr.value.(stateMachineIllegalStatePanic)
+	if !ok {
+		return nil, false
+	}
+
+	return panicErr, true
+}
+
 func (wth *workflowTaskHandlerImpl) completeWorkflow(
 	eventHandler *workflowExecutionEventHandlerImpl,
 	task *s.PollForDecisionTaskResponse,
@@ -1312,7 +1322,7 @@ func (wth *workflowTaskHandlerImpl) completeWorkflow(
 
 	if closeDecision != nil {
 		decisions = append(decisions, closeDecision)
-		elapsed := time.Now().Sub(workflowContext.workflowStartTime)
+		elapsed := time.Since(workflowContext.workflowStartTime)
 		metricsScope.Timer(metrics.WorkflowEndToEndLatency).Record(elapsed)
 		forceNewDecision = false
 	}
@@ -1845,7 +1855,8 @@ func traceLog(fn func()) {
 	}
 }
 
-func workflowCategorizedByTimeout(executionTimeout int32) string {
+func workflowCategorizedByTimeout(wfContext *workflowExecutionContextImpl) string {
+	executionTimeout := wfContext.workflowInfo.ExecutionStartToCloseTimeoutSeconds
 	if executionTimeout <= defaultInstantLivedWorkflowTimeoutUpperLimitInSec {
 		return "instant"
 	} else if executionTimeout <= defaultShortLivedWorkflowTimeoutUpperLimitInSec {

internal/internal_task_handlers_test.go

@@ -869,28 +869,20 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicLogNonexistingI
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, t.registry)
 	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 
-	t.NotNil(request)
-	response := request.(*s.RespondDecisionTaskFailedRequest)
-
-	// NOTE: we might acctually want to return an error
-	// but since previously we checked the wrong error type, it may break existing customers workflow
-	// The issue is that we change the error type and that we change the error message, the customers
-	// are checking the error string - we plan to wrap all errors to avoid this issue in client v2
-	t.NoError(err)
-	t.NotNil(response)
+	t.Nil(request)
+	t.ErrorContains(err, "nondeterministic workflow")
 
 	// Check that the error was logged
-	ignoredWorkflowLogs := logs.FilterMessage("Ignored workflow panic error")
-	require.Len(t.T(), ignoredWorkflowLogs.All(), 1)
+	illegalPanicLogs := logs.FilterMessage("Illegal state caused panic")
+	require.Len(t.T(), illegalPanicLogs.All(), 1)
 
-	replayErrorField := findLogField(ignoredWorkflowLogs.All()[0], "ReplayError")
+	replayErrorField := findLogField(illegalPanicLogs.All()[0], "ReplayError")
 	require.NotNil(t.T(), replayErrorField)
 	require.Equal(t.T(), zapcore.ErrorType, replayErrorField.Type)
 	require.ErrorContains(t.T(), replayErrorField.Interface.(error),
 		"nondeterministic workflow: "+
 			"history event is ActivityTaskScheduled: (ActivityId:NotAnActivityID, ActivityType:(Name:pkg.Greeter_Activity), TaskList:(Name:taskList), Input:[]), "+
 			"replay decision is ScheduleActivityTask: (ActivityId:0, ActivityType:(Name:Greeter_Activity), TaskList:(Name:taskList)")
-
 }
 
 func (t *TaskHandlersTestSuite) TestWorkflowTask_WorkflowReturnsPanicError() {

test/activity_test.go

@@ -105,9 +105,7 @@ func (a *Activities) invoked() []string {
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	result := make([]string, len(a.invocations))
-	for i := range a.invocations {
-		result[i] = a.invocations[i]
-	}
+	copy(result, a.invocations)
 	return result
 }
 

test/integration_test.go

@@ -23,6 +23,7 @@ package test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"strings"
@@ -148,18 +149,24 @@ func (ts *IntegrationTestSuite) SetupTest() {
 	ts.seq++
 	ts.activities.clearInvoked()
 	ts.taskListName = fmt.Sprintf("tl-%v", ts.seq)
-	ts.worker = worker.New(ts.rpcClient.Interface, domainName, ts.taskListName, worker.Options{
-		DisableStickyExecution: ts.config.IsStickyOff,
-		Logger:                 zaptest.NewLogger(ts.T()),
-		ContextPropagators:     []workflow.ContextPropagator{NewStringMapPropagator([]string{testContextKey})},
-	})
 	ts.tracer = newtracingInterceptorFactory()
+}
+
+func (ts *IntegrationTestSuite) BeforeTest(suiteName, testName string) {
 	options := worker.Options{
 		DisableStickyExecution:            ts.config.IsStickyOff,
 		Logger:                            zaptest.NewLogger(ts.T()),
 		WorkflowInterceptorChainFactories: []interceptors.WorkflowInterceptorFactory{ts.tracer},
 		ContextPropagators:                []workflow.ContextPropagator{NewStringMapPropagator([]string{testContextKey})},
 	}
+
+	if testName == "TestNonDeterministicWorkflowQuery" || testName == "TestNonDeterministicWorkflowFailPolicy" {
+		options.NonDeterministicWorkflowPolicy = worker.NonDeterministicWorkflowPolicyFailWorkflow
+
+		// disable sticky executon so each workflow yield will require rerunning it from beginning
+		options.DisableStickyExecution = true
+	}
+
 	ts.worker = worker.New(ts.rpcClient.Interface, domainName, ts.taskListName, options)
 	ts.registerWorkflowsAndActivities(ts.worker)
 	ts.Nil(ts.worker.Start())
@@ -270,7 +277,7 @@ func (ts *IntegrationTestSuite) TestStackTraceQuery() {
 	ts.NoError(err)
 	ts.NotNil(value)
 	var trace string
-	ts.Nil(value.Get(&trace))
+	ts.NoError(value.Get(&trace))
 	ts.True(strings.Contains(trace, "go.uber.org/cadence/test.(*Workflows).Basic"))
 }
 
@@ -303,7 +310,7 @@ func (ts *IntegrationTestSuite) TestConsistentQuery() {
 	ts.NotNil(value.QueryResult)
 	ts.Nil(value.QueryRejected)
 	var queryResult string
-	ts.Nil(value.QueryResult.Get(&queryResult))
+	ts.NoError(value.QueryResult.Get(&queryResult))
 	ts.Equal("signal-input", queryResult)
 }
 
@@ -428,7 +435,6 @@ func (ts *IntegrationTestSuite) TestChildWFWithParentClosePolicyTerminate() {
 	resp, err := ts.libClient.DescribeWorkflowExecution(context.Background(), childWorkflowID, "")
 	ts.NoError(err)
 	ts.True(resp.WorkflowExecutionInfo.GetCloseTime() > 0)
-
 }
 
 func (ts *IntegrationTestSuite) TestChildWFWithParentClosePolicyAbandon() {
@@ -437,7 +443,7 @@ func (ts *IntegrationTestSuite) TestChildWFWithParentClosePolicyAbandon() {
 	ts.NoError(err)
 	resp, err := ts.libClient.DescribeWorkflowExecution(context.Background(), childWorkflowID, "")
 	ts.NoError(err)
-	ts.True(resp.WorkflowExecutionInfo.GetCloseTime() == 0)
+	ts.Zerof(resp.WorkflowExecutionInfo.GetCloseTime(), "Expected close time to be zero, got %d. Describe response: %#v", resp.WorkflowExecutionInfo.GetCloseTime(), resp)
 }
 
 func (ts *IntegrationTestSuite) TestChildWFCancel() {
@@ -512,6 +518,33 @@ func (ts *IntegrationTestSuite) TestDomainUpdate() {
 	ts.Equal(description, *domain.DomainInfo.Description)
 }
 
+func (ts *IntegrationTestSuite) TestNonDeterministicWorkflowFailPolicy() {
+	err := ts.executeWorkflow("test-nondeterminism-failpolicy", ts.workflows.NonDeterminismSimulatorWorkflow, nil)
+	var customErr *internal.CustomError
+	ok := errors.As(err, &customErr)
+	ts.Truef(ok, "expected CustomError but got %T", err)
+	ts.Equal("NonDeterministicWorkflowPolicyFailWorkflow", customErr.Reason())
+}
+
+func (ts *IntegrationTestSuite) TestNonDeterministicWorkflowQuery() {
+	ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout)
+	defer cancel()
+	run, err := ts.libClient.ExecuteWorkflow(ctx, ts.startWorkflowOptions("test-nondeterministic-query"), ts.workflows.NonDeterminismSimulatorWorkflow)
+	ts.Nil(err)
+	err = run.Get(ctx, nil)
+	var customErr *internal.CustomError
+	ok := errors.As(err, &customErr)
+	ts.Truef(ok, "expected CustomError but got %T", err)
+	ts.Equal("NonDeterministicWorkflowPolicyFailWorkflow", customErr.Reason())
+
+	// query failed workflow should still work
+	value, err := ts.libClient.QueryWorkflow(ctx, "test-nondeterministic-query", run.GetRunID(), "__stack_trace")
+	ts.NoError(err)
+	ts.NotNil(value)
+	var trace string
+	ts.NoError(value.Get(&trace))
+}
+
 func (ts *IntegrationTestSuite) registerDomain() {
 	ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout)
 	defer cancel()