Ignore workflowPanicError in query task and add warn logs (#1094)

Author: yycptt

internal/internal_event_handlers.go

@@ -781,8 +781,8 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 			topLine := fmt.Sprintf("process event for %s [panic]:", weh.workflowInfo.TaskListName)
 			st := getStackTraceRaw(topLine, 7, 0)
 			weh.logger.Error("ProcessEvent panic.",
-				zap.String("PanicError", fmt.Sprintf("%v", p)),
-				zap.String("PanicStack", st))
+				zap.String(tagPanicError, fmt.Sprintf("%v", p)),
+				zap.String(tagPanicStack, st))
 
 			weh.Complete(nil, newWorkflowPanicError(p, st))
 		}

internal/internal_logging_tags.go

@@ -38,4 +38,6 @@ const (
 	tagLocalActivityType = "LocalActivityType"
 	tagQueryType         = "QueryType"
 	tagVisibilityQuery   = "VisibilityQuery"
+	tagPanicError        = "PanicError"
+	tagPanicStack        = "PanicStack"
 )

internal/internal_task_handlers.go

@@ -823,6 +823,13 @@ func (w *workflowExecutionContextImpl) ProcessWorkflowTask(workflowTask *workflo
 
 	skipReplayCheck := w.skipReplayCheck()
 	isReplayTest := task.GetPreviousStartedEventId() == replayPreviousStartedEventID
+	if isReplayTest {
+		w.wth.logger.Info("Processing workflow task in replay test mode",
+			zap.String(tagWorkflowType, task.WorkflowType.GetName()),
+			zap.String(tagWorkflowID, task.WorkflowExecution.GetWorkflowId()),
+			zap.String(tagRunID, task.WorkflowExecution.GetRunId()),
+		)
+	}
 	// Process events
 ProcessEvents:
 	for {
@@ -917,12 +924,19 @@ ProcessEvents:
 		}
 	}
 	if nonDeterministicErr == nil && w.err != nil {
-		if isReplayTest {
-			// NOTE: we should check the following error regardless if it's in replay test or not
-			// but since we are not checking it previously, it may break existing customers workflow
-			if panicErr, ok := w.err.(*workflowPanicError); ok && panicErr.value != nil {
-				if _, isStateMachinePanic := panicErr.value.(stateMachineIllegalStatePanic); isStateMachinePanic {
+		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
 					nonDeterministicErr = panicErr
+				} else {
+					w.wth.logger.Warn("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),
+					)
 				}
 			}
 		}
@@ -1455,12 +1469,32 @@ func (wth *workflowTaskHandlerImpl) completeWorkflow(
 	// for query task
 	if task.Query != nil {
 		queryCompletedRequest := &s.RespondQueryTaskCompletedRequest{TaskToken: task.TaskToken}
-		if panicErr, ok := workflowContext.err.(*workflowPanicError); ok {
+		if panicErr, ok := workflowContext.err.(*PanicError); ok {
+			// NOTE: this code path should never be executed, we should check for workflowPanicError instead of PanicError
+			wth.logger.Warn("Encountered PanicError in workflow query task",
+				zap.String(tagWorkflowID, task.WorkflowExecution.GetWorkflowId()),
+				zap.String(tagRunID, task.WorkflowExecution.GetRunId()),
+				zap.String(tagPanicError, panicErr.Error()),
+				zap.String(tagPanicStack, panicErr.StackTrace()),
+			)
+
 			queryCompletedRequest.CompletedType = common.QueryTaskCompletedTypePtr(s.QueryTaskCompletedTypeFailed)
 			queryCompletedRequest.ErrorMessage = common.StringPtr("Workflow panic: " + panicErr.Error())
 			return queryCompletedRequest
 		}
 
+		if workflowPanicErr, ok := workflowContext.err.(*workflowPanicError); ok {
+			// NOTE: in this case we should return complete query task with CompletedTypeFailed
+			// but we didn't check for the right error type before, this may break existing customer
+			wth.logger.Warn("Ignored workflow panic error for query, query result may be partial",
+				zap.String(tagWorkflowID, task.WorkflowExecution.GetWorkflowId()),
+				zap.String(tagRunID, task.WorkflowExecution.GetRunId()),
+				zap.String(tagPanicError, workflowPanicErr.Error()),
+				zap.String(tagPanicStack, workflowPanicErr.StackTrace()),
+				zap.Int64("PreviousStartedEventID", task.GetPreviousStartedEventId()),
+			)
+		}
+
 		result, err := eventHandler.ProcessQuery(task.Query.GetQueryType(), task.Query.QueryArgs)
 		if err != nil {
 			queryCompletedRequest.CompletedType = common.QueryTaskCompletedTypePtr(s.QueryTaskCompletedTypeFailed)
@@ -1481,8 +1515,8 @@ func (wth *workflowTaskHandlerImpl) completeWorkflow(
 		wth.logger.Error("Workflow panic.",
 			zap.String(tagWorkflowID, task.WorkflowExecution.GetWorkflowId()),
 			zap.String(tagRunID, task.WorkflowExecution.GetRunId()),
-			zap.String("PanicError", panicErr.Error()),
-			zap.String("PanicStack", panicErr.StackTrace()))
+			zap.String(tagPanicError, panicErr.Error()),
+			zap.String(tagPanicStack, panicErr.StackTrace()))
 		return errorToFailDecisionTask(task.TaskToken, panicErr, wth.identity)
 	}
 
@@ -1840,8 +1874,8 @@ func (ath *activityTaskHandlerImpl) Execute(taskList string, t *s.PollForActivit
 				zap.String(tagWorkflowID, t.WorkflowExecution.GetWorkflowId()),
 				zap.String(tagRunID, t.WorkflowExecution.GetRunId()),
 				zap.String(tagActivityType, activityType),
-				zap.String("PanicError", fmt.Sprintf("%v", p)),
-				zap.String("PanicStack", st))
+				zap.String(tagPanicError, fmt.Sprintf("%v", p)),
+				zap.String(tagPanicStack, st))
 			metricsScope.Counter(metrics.ActivityTaskPanicCounter).Inc(1)
 			panicErr := newPanicError(p, st)
 			result, err = convertActivityResultToRespondRequest(ath.identity, t.TaskToken, nil, panicErr, ath.dataConverter), nil

internal/internal_task_pollers.go

@@ -603,8 +603,8 @@ func (lath *localActivityTaskHandler) executeLocalActivityTask(task *localActivi
 				zap.String(tagWorkflowID, task.params.WorkflowInfo.WorkflowExecution.ID),
 				zap.String(tagRunID, task.params.WorkflowInfo.WorkflowExecution.RunID),
 				zap.String(tagActivityType, activityType),
-				zap.String("PanicError", fmt.Sprintf("%v", p)),
-				zap.String("PanicStack", st))
+				zap.String(tagPanicError, fmt.Sprintf("%v", p)),
+				zap.String(tagPanicStack, st))
 			metricsScope.Counter(metrics.LocalActivityPanicCounter).Inc(1)
 			panicErr := newPanicError(p, st)
 			result = &localActivityResult{

internal/internal_worker_base.go

@@ -313,8 +313,8 @@ func (bw *baseWorker) processTask(task interface{}) {
 			topLine := fmt.Sprintf("base worker for %s [panic]:", bw.options.workerType)
 			st := getStackTraceRaw(topLine, 7, 0)
 			bw.logger.Error("Unhandled panic.",
-				zap.String("PanicError", fmt.Sprintf("%v", p)),
-				zap.String("PanicStack", st))
+				zap.String(tagPanicError, fmt.Sprintf("%v", p)),
+				zap.String(tagPanicStack, st))
 		}
 
 		if isPolledTask {