Force terminate workflow execution when exceeds system limit

Moved the logic to force terminate workflow execution to
workflowExecutionContext to make sure it is triggered for all cases
when workflow execution exceeds size limit.

Previously we are only performing this check on workflow task
completion.  This could result in scenarios where workflow would
never be forced terminated.

Removed the logic to fail the workflow on command processing as
it is redundant after moving force termination logic to more generic
location.

Author: samarabbas

host/sizelimit_test.go

@@ -36,6 +36,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/serviceerror"
 
 	commandpb "go.temporal.io/api/command/v1"
 	commonpb "go.temporal.io/api/common/v1"
@@ -44,6 +45,7 @@ import (
 	taskqueuepb "go.temporal.io/api/taskqueue/v1"
 	"go.temporal.io/api/workflowservice/v1"
 
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/payloads"
 	"go.temporal.io/server/common/primitives/timestamp"
@@ -153,19 +155,55 @@ func (s *sizeLimitIntegrationSuite) TestTerminateWorkflowCausedBySizeLimit() {
 	}
 
 	for i := int32(0); i < activityCount-1; i++ {
-		_, err := poller.PollAndProcessWorkflowTask(false, false)
-		s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
+		dwResp, err := s.engine.DescribeWorkflowExecution(NewContext(), &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: s.namespace,
+			Execution: &commonpb.WorkflowExecution{
+				WorkflowId: id,
+				RunId:      we.RunId,
+			},
+		})
 		s.NoError(err)
 
-		err = poller.PollAndProcessActivityTask(false)
-		s.Logger.Info("PollAndProcessActivityTask", tag.Error(err))
-		s.NoError(err)
+		// Poll workflow task only if it is running
+		if dwResp.WorkflowExecutionInfo.Status == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
+			_, err := poller.PollAndProcessWorkflowTask(false, false)
+			s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
+			s.NoError(err)
+
+			err = poller.PollAndProcessActivityTask(false)
+			s.Logger.Info("PollAndProcessActivityTask", tag.Error(err))
+			s.NoError(err)
+		}
 	}
 
-	// process this workflow task will trigger history exceed limit error
-	_, err := poller.PollAndProcessWorkflowTask(false, false)
-	s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
-	s.NoError(err)
+	var signalErr error
+	// Send signals until workflow is force terminated
+SignalLoop:
+	for i := 0; i < 10; i++ {
+		// Send another signal without RunID
+		signalName := "another signal"
+		signalInput := payloads.EncodeString("another signal input")
+		_, signalErr = s.engine.SignalWorkflowExecution(NewContext(), &workflowservice.SignalWorkflowExecutionRequest{
+			Namespace: s.namespace,
+			WorkflowExecution: &commonpb.WorkflowExecution{
+				WorkflowId: id,
+			},
+			SignalName: signalName,
+			Input:      signalInput,
+			Identity:   identity,
+		})
+
+		if signalErr != nil {
+			break SignalLoop
+		}
+	}
+	s.EqualError(signalErr, common.FailureReasonSizeExceedsLimit)
+	s.IsType(&serviceerror.InvalidArgument{}, signalErr)
+
+	s.printWorkflowHistory(s.namespace, &commonpb.WorkflowExecution{
+		WorkflowId: id,
+		RunId:      we.GetRunId(),
+	})
 
 	// verify last event is terminated event
 	historyResponse, err := s.engine.GetWorkflowExecutionHistory(NewContext(), &workflowservice.GetWorkflowExecutionHistoryRequest{
@@ -178,9 +216,7 @@ func (s *sizeLimitIntegrationSuite) TestTerminateWorkflowCausedBySizeLimit() {
 	s.NoError(err)
 	history := historyResponse.History
 	lastEvent := history.Events[len(history.Events)-1]
-	s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED, lastEvent.GetEventType())
-	failedEventAttributes := lastEvent.GetWorkflowExecutionFailedEventAttributes()
-	s.True(failedEventAttributes.GetFailure().GetServerFailureInfo().GetNonRetryable())
+	s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED, lastEvent.GetEventType())
 
 	// verify visibility is correctly processed from open to close
 	isCloseCorrect := false

host/taskpoller.go

@@ -152,6 +152,10 @@ Loop:
 			Identity:  p.Identity,
 		})
 
+		if common.IsServiceNonRetryableError(err1) {
+			return false, nil, err1
+		}
+
 		if err1 == history.ErrDuplicate {
 			p.Logger.Info("Duplicate Workflow task: Polling again")
 			continue Loop

service/history/commandChecker.go

@@ -164,43 +164,6 @@ func (c *workflowSizeChecker) failWorkflowIfPayloadSizeExceedsLimit(
 	return true, nil
 }
 
-func (c *workflowSizeChecker) failWorkflowSizeExceedsLimit() (bool, error) {
-	historyCount := int(c.mutableState.GetNextEventID()) - 1
-	historySize := int(c.executionStats.HistorySize)
-
-	if historySize > c.historySizeLimitError || historyCount > c.historyCountLimitError {
-		executionInfo := c.mutableState.GetExecutionInfo()
-		c.logger.Error("history size exceeds error limit.",
-			tag.WorkflowNamespaceID(executionInfo.NamespaceId),
-			tag.WorkflowID(executionInfo.WorkflowId),
-			tag.WorkflowRunID(executionInfo.ExecutionState.RunId),
-			tag.WorkflowHistorySize(historySize),
-			tag.WorkflowEventCount(historyCount))
-
-		attributes := &commandpb.FailWorkflowExecutionCommandAttributes{
-			Failure: failure.NewServerFailure(common.FailureReasonSizeExceedsLimit, true),
-		}
-
-		if _, err := c.mutableState.AddFailWorkflowEvent(c.completedID, enumspb.RETRY_STATE_NON_RETRYABLE_FAILURE, attributes); err != nil {
-			return false, err
-		}
-		return true, nil
-	}
-
-	if historySize > c.historySizeLimitWarn || historyCount > c.historyCountLimitWarn {
-		executionInfo := c.mutableState.GetExecutionInfo()
-		c.logger.Warn("history size exceeds warn limit.",
-			tag.WorkflowNamespaceID(executionInfo.NamespaceId),
-			tag.WorkflowID(executionInfo.WorkflowId),
-			tag.WorkflowRunID(executionInfo.ExecutionState.RunId),
-			tag.WorkflowHistorySize(historySize),
-			tag.WorkflowEventCount(historyCount))
-		return false, nil
-	}
-
-	return false, nil
-}
-
 func (v *commandAttrValidator) validateActivityScheduleAttributes(
 	namespaceID string,
 	targetNamespaceID string,

service/history/historyEngine.go

@@ -196,6 +196,8 @@ var (
 	ErrConsistentQueryBufferExceeded = serviceerror.NewInternal("consistent query buffer is full, cannot accept new consistent queries")
 	// ErrEmptyHistoryRawEventBatch indicate that one single batch of history raw events is of size 0
 	ErrEmptyHistoryRawEventBatch = serviceerror.NewInvalidArgument("encounter empty history batch")
+	// ErrSizeExceedsLimit is error indicating workflow execution has exceeded system defined limit
+	ErrSizeExceedsLimit = serviceerror.NewInvalidArgument(common.FailureReasonSizeExceedsLimit)
 
 	// FailedWorkflowStatuses is a set of failed workflow close states, used for start workflow policy
 	// for start workflow execution API

service/history/workflowExecutionContext.go

@@ -138,6 +138,7 @@ type (
 		logger            log.Logger
 		metricsClient     metrics.Client
 		timeSource        clock.TimeSource
+		config            *Config
 
 		mutex           locks.Mutex
 		mutableState    mutableState
@@ -168,6 +169,7 @@ func newWorkflowExecutionContext(
 		logger:            logger,
 		metricsClient:     shard.GetMetricsClient(),
 		timeSource:        shard.GetTimeSource(),
+		config:            shard.GetConfig(),
 		mutex:             locks.NewMutex(),
 		stats: &persistenceblobs.ExecutionStats{
 			HistorySize: 0,
@@ -572,14 +574,30 @@ func (c *workflowExecutionContextImpl) updateWorkflowExecutionAsActive(
 	now time.Time,
 ) error {
 
-	return c.updateWorkflowExecutionWithNew(
+	// We only perform this check on active cluster for the namespace
+	forceTerminate, err := c.enforceSizeCheck()
+	if err != nil {
+		return err
+	}
+
+	if err := c.updateWorkflowExecutionWithNew(
 		now,
 		persistence.UpdateWorkflowModeUpdateCurrent,
 		nil,
 		nil,
 		transactionPolicyActive,
 		nil,
-	)
+	); err != nil {
+		return err
+	}
+
+	if forceTerminate {
+		// Returning an error which should be not be retried in most of the cases
+		// Caller can explicitly check for this error if they need to special case on force termination of workflow
+		return ErrSizeExceedsLimit
+	}
+
+	return nil
 }
 
 func (c *workflowExecutionContextImpl) updateWorkflowExecutionWithNewAsActive(
@@ -1146,3 +1164,62 @@ func (c *workflowExecutionContextImpl) reapplyEvents(
 
 	return err
 }
+
+// Returns true if execution is forced terminated
+func (c *workflowExecutionContextImpl) enforceSizeCheck() (bool, error) {
+	historySizeLimitWarn := c.config.HistorySizeLimitWarn(c.getNamespace())
+	historySizeLimitError := c.config.HistorySizeLimitError(c.getNamespace())
+	historyCountLimitWarn := c.config.HistoryCountLimitWarn(c.getNamespace())
+	historyCountLimitError := c.config.HistoryCountLimitError(c.getNamespace())
+
+	historySize := int(c.stats.HistorySize)
+	historyCount := int(c.mutableState.GetNextEventID() - 1)
+
+	// Hard terminate workflow if still running and breached size or count limit
+	if (historySize > historySizeLimitError || historyCount > historyCountLimitError) &&
+		c.mutableState.IsWorkflowExecutionRunning() {
+		c.logger.Error("history size exceeds error limit.",
+			tag.WorkflowNamespaceID(c.namespaceID),
+			tag.WorkflowID(c.workflowExecution.GetWorkflowId()),
+			tag.WorkflowRunID(c.workflowExecution.GetRunId()),
+			tag.WorkflowHistorySize(historySize),
+			tag.WorkflowEventCount(historyCount))
+
+		// Discard pending changes in mutableState so we can apply terminate state transition
+		c.clear()
+
+		// Reload muutable state
+		mutableState, err := c.loadWorkflowExecution()
+		if err != nil {
+			return false, err
+		}
+
+		// Terminate workflow is written as a separate batch and might result in more than one event as we close the
+		// outstanding workflow task before terminating the workflow
+		eventBatchFirstEventID := mutableState.GetNextEventID()
+		if err := terminateWorkflow(
+			mutableState,
+			eventBatchFirstEventID,
+			common.FailureReasonSizeExceedsLimit,
+			nil,
+			identityHistoryService,
+		); err != nil {
+			return false, err
+		}
+
+		// Return true to caller to indicate workflow is forced terminated
+		return true, nil
+	}
+
+	if historySize > historySizeLimitWarn || historyCount > historyCountLimitWarn {
+		executionInfo := c.mutableState.GetExecutionInfo()
+		c.logger.Warn("history size exceeds warn limit.",
+			tag.WorkflowNamespaceID(executionInfo.NamespaceId),
+			tag.WorkflowID(executionInfo.WorkflowId),
+			tag.WorkflowRunID(executionInfo.ExecutionState.RunId),
+			tag.WorkflowHistorySize(historySize),
+			tag.WorkflowEventCount(historyCount))
+	}
+
+	return false, nil
+}

service/history/workflowTaskHandler.go

@@ -124,15 +124,9 @@ func (handler *workflowTaskHandlerImpl) handleCommands(
 	commands []*commandpb.Command,
 ) error {
 
-	// overall workflow size / count check
-	failWorkflow, err := handler.sizeLimitChecker.failWorkflowSizeExceedsLimit()
-	if err != nil || failWorkflow {
-		return err
-	}
-
 	for _, command := range commands {
 
-		err = handler.handleCommand(command)
+		err := handler.handleCommand(command)
 		if err != nil || handler.stopProcessing {
 			return err
 		}