Improve workflow replayer (#1082)

- Catch WorkflowPanicError instead of PanicError when processing workflow tasks
- Check if the last event is ContinueAsNew when replay result is workflow completed
- Correctly handle decision & event match check when replaying partial workflow history
- Add support for data converter, context propagator, tracer, interceptor in workflow replayer and shadower.
- Fetch next history page if possible

Author: yycptt

internal/internal_decision_state_machine_test.go

@@ -624,13 +624,13 @@ func Test_CancelExternalWorkflowStateMachine_Failed(t *testing.T) {
 	require.NotNil(t, err)
 }
 
-func runAndCatchPanic(f func()) (err *PanicError) {
+func runAndCatchPanic(f func()) (err *workflowPanicError) {
 	// panic handler
 	defer func() {
 		if p := recover(); p != nil {
 			topLine := "runAndCatchPanic [panic]:"
 			st := getStackTraceRaw(topLine, 7, 0)
-			err = newPanicError(p, st) // Fail decision on panic
+			err = newWorkflowPanicError(p, st) // Fail decision on panic
 		}
 	}()
 

internal/internal_task_handlers.go

@@ -268,6 +268,10 @@ func (eh *history) NextDecisionEvents() (result []*s.HistoryEvent, markers []*s.
 	return result, markers, checksum, err
 }
 
+func (eh *history) HasNextDecisionEvents() bool {
+	return len(eh.next) != 0 || eh.currentIndex != len(eh.loadedEvents) || eh.hasMoreEvents()
+}
+
 func (eh *history) hasMoreEvents() bool {
 	historyIterator := eh.workflowTask.historyIterator
 	return historyIterator != nil && historyIterator.HasNextPage()
@@ -468,7 +472,7 @@ func (w *workflowExecutionContextImpl) Lock() {
 
 func (w *workflowExecutionContextImpl) Unlock(err error) {
 	if err != nil || w.err != nil || w.isWorkflowCompleted || (w.wth.disableStickyExecution && !w.hasPendingLocalActivityWork()) {
-		// TODO: in case of closed, it asumes the close decision always succeed. need server side change to return
+		// TODO: in case of closed, it assumes the close decision always succeed. need server side change to return
 		// error to indicate the close failure case. This should be rear case. For now, always remove the cache, and
 		// if the close decision failed, the next decision will have to rebuild the state.
 		if getWorkflowCache().Exist(w.workflowInfo.WorkflowExecution.RunID) {
@@ -818,6 +822,7 @@ func (w *workflowExecutionContextImpl) ProcessWorkflowTask(workflowTask *workflo
 	var respondEvents []*s.HistoryEvent
 
 	skipReplayCheck := w.skipReplayCheck()
+	isReplayTest := task.GetPreviousStartedEventId() == replayPreviousStartedEventID
 	// Process events
 ProcessEvents:
 	for {
@@ -888,7 +893,8 @@ ProcessEvents:
 			}
 		}
 		isReplay := len(reorderedEvents) > 0 && reorderedHistory.IsReplayEvent(reorderedEvents[len(reorderedEvents)-1])
-		if isReplay {
+		lastDecisionEventsForReplayTest := isReplayTest && !reorderedHistory.HasNextDecisionEvents()
+		if isReplay && !lastDecisionEventsForReplayTest {
 			eventDecisions := eventHandler.decisionsHelper.getDecisions(true)
 			if len(eventDecisions) > 0 && !skipReplayCheck {
 				replayDecisions = append(replayDecisions, eventDecisions...)
@@ -904,14 +910,14 @@ 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
-	if !skipReplayCheck && !w.isWorkflowCompleted {
+	if !skipReplayCheck && !w.isWorkflowCompleted || isReplayTest {
 		// check if decisions from reply matches to the history events
 		if err := matchReplayWithHistory(replayDecisions, respondEvents); err != nil {
 			nonDeterministicErr = err
 		}
 	}
 	if nonDeterministicErr == nil && w.err != nil {
-		if panicErr, ok := w.err.(*PanicError); ok && panicErr.value != nil {
+		if panicErr, ok := w.err.(*workflowPanicError); ok && panicErr.value != nil {
 			if _, isStateMachinePanic := panicErr.value.(stateMachineIllegalStatePanic); isStateMachinePanic {
 				nonDeterministicErr = panicErr
 			}
@@ -1445,7 +1451,7 @@ func (wth *workflowTaskHandlerImpl) completeWorkflow(
 	// for query task
 	if task.Query != nil {
 		queryCompletedRequest := &s.RespondQueryTaskCompletedRequest{TaskToken: task.TaskToken}
-		if panicErr, ok := workflowContext.err.(*PanicError); ok {
+		if panicErr, ok := workflowContext.err.(*workflowPanicError); ok {
 			queryCompletedRequest.CompletedType = common.QueryTaskCompletedTypePtr(s.QueryTaskCompletedTypeFailed)
 			queryCompletedRequest.ErrorMessage = common.StringPtr("Workflow panic: " + panicErr.Error())
 			return queryCompletedRequest

internal/internal_task_handlers_test.go

@@ -132,6 +132,10 @@ func createTestEventWorkflowExecutionCompleted(eventID int64, attr *s.WorkflowEx
 	return &s.HistoryEvent{EventId: common.Int64Ptr(eventID), EventType: common.EventTypePtr(s.EventTypeWorkflowExecutionCompleted), WorkflowExecutionCompletedEventAttributes: attr}
 }
 
+func createTestEventWorkflowExecutionContinuedAsNew(eventID int64, attr *s.WorkflowExecutionContinuedAsNewEventAttributes) *s.HistoryEvent {
+	return &s.HistoryEvent{EventId: common.Int64Ptr(eventID), EventType: common.EventTypePtr(s.EventTypeWorkflowExecutionContinuedAsNew), WorkflowExecutionContinuedAsNewEventAttributes: attr}
+}
+
 func createTestEventWorkflowExecutionStarted(eventID int64, attr *s.WorkflowExecutionStartedEventAttributes) *s.HistoryEvent {
 	return &s.HistoryEvent{EventId: common.Int64Ptr(eventID), EventType: common.EventTypePtr(s.EventTypeWorkflowExecutionStarted), WorkflowExecutionStartedEventAttributes: attr}
 }
@@ -629,8 +633,7 @@ func (t *TaskHandlersTestSuite) TestWithTruncatedHistory() {
 		previousStartedEventID int64
 		isResultErr            bool
 	}{
-		{0, 0, false},
-		{0, 3, false},
+		{0, 6, false},
 		{10, 0, true},
 		{10, 6, true},
 	}
@@ -648,7 +651,7 @@ func (t *TaskHandlersTestSuite) TestWithTruncatedHistory() {
 			t.Error(err, "testcase %v failed", i)
 			t.Nil(request)
 			t.Contains(err.Error(), "premature end of stream")
-			t.EqualValues(getWorkflowCache().Size(), 0)
+			t.EqualValues(getWorkflowCache().Size(), 1)
 			continue
 		}
 

internal/worker.go

@@ -184,7 +184,7 @@ type (
 		// default: false
 		EnableSessionWorker bool
 
-		// Uncomment this option when we support automatic restablish failed sessions.
+		// Uncomment this option when we support automatic reestablish failed sessions.
 		// Optional: The identifier of the resource consumed by sessions.
 		// It's the user's responsibility to ensure there's only one worker using this resourceID.
 		// For now, if user doesn't specify one, a new uuid will be used as the resourceID.
@@ -207,8 +207,14 @@ type (
 		Tracer opentracing.Tracer
 
 		// Optional: Enable worker for running shadowing workflows to replay existing workflows
-		// If set to true, worker will run in shadow mode and all other workers (decision, activity, session)
+		// If set to true:
+		// 1. Worker will run in shadow mode and all other workers (decision, activity, session)
 		// will be disabled to prevent them from updating existing workflow states.
+		// 2. DataConverter, WorkflowInterceptorChainFactories, ContextPropagators, Tracer will be
+		// used as ReplayOptions and forwarded to the underlying WorkflowReplayer.
+		// The actual shadower activity worker will not use them.
+		// 3. TaskList will become Domain-TaskList, to prevent conflict across domains as there's
+		// only one shadowing domain which is responsible for shadowing workflows for all domains.
 		// default: false
 		EnableShadowWorker bool
 
@@ -236,16 +242,8 @@ const (
 	// Whereas default does *NOT* reply anything back to the server, fail workflow replies back with a request
 	// to fail the workflow execution.
 	NonDeterministicWorkflowPolicyFailWorkflow
-
-	// ReplayDomainName is domainName for replay because startEvent doesn't contain it
-	ReplayDomainName = "ReplayDomain"
 )
 
-// IsReplayDomain checks if the domainName is from replay
-func IsReplayDomain(dn string) bool {
-	return ReplayDomainName == dn
-}
-
 // 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.