feat(diagnostics): diagnose large workflow histories (#8490)

## What changed?
The diagnostics failure invariant now recognizes workflows that failed
with the "history exceeds limit" error and reports a dedicated root
cause, including the history event count of the failed workflow.

Misc:
- The variable `FailureReasonSizeExceedsLimit` has been renamed to
`FailureReasonHistorySizeExceedsLimit`

## Why?
When a workflow fails because its history grew past the size or event
count limit, diagnostics previously bucketed it as a generic custom
error ("Customised error returned by the activity/workflow"), which
points users in the wrong direction. The new root cause names the actual
problem. The runbook update linked below suggests remediation
(ContinueAsNew, smaller payloads, avoiding unbounded loops in one run).

## How did you test it?
```
make pr GEN_DIR=service/worker/diagnostics  # clean, no generated diffs
make build
make test                                   # all unit tests pass
go test -race ./service/worker/diagnostics/...
```

## Potential risks
None — isolated change to the diagnostics worker; no API/schema/flag
impact.

## Release notes
Workflow diagnostics now identifies failures caused by exceeding history
size/count limits and surfaces an actionable root cause.

## Documentation Changes
Failures runbook update:
cadence-workflow/Cadence-Docs#406

---------

Signed-off-by: Adhitya Mamallan <adhitya.mamallan@uber.com>

Author: adhityamamallan

common/util.go

@@ -103,8 +103,8 @@ const (
 	FailureReasonHeartbeatExceedsLimit = "HEARTBEAT_EXCEEDS_LIMIT"
 	// FailureReasonDecisionBlobSizeExceedsLimit is the failureReason for decision blob exceeds size limit
 	FailureReasonDecisionBlobSizeExceedsLimit = "DECISION_BLOB_SIZE_EXCEEDS_LIMIT"
-	// FailureReasonSizeExceedsLimit is reason to fail workflow when history size or count exceed limit
-	FailureReasonSizeExceedsLimit = "HISTORY_EXCEEDS_LIMIT"
+	// FailureReasonHistorySizeExceedsLimit is reason to fail workflow when history size or count exceed limit
+	FailureReasonHistorySizeExceedsLimit = "HISTORY_EXCEEDS_LIMIT"
 	// FailureReasonTransactionSizeExceedsLimit is the failureReason for when transaction cannot be committed because it exceeds size limit
 	FailureReasonTransactionSizeExceedsLimit = "TRANSACTION_SIZE_EXCEEDS_LIMIT"
 	// FailureReasonDecisionAttemptsExceedsLimit is reason to fail workflow when decision attempts fail too many times

host/size_limit_test.go

@@ -180,7 +180,7 @@ func (s *SizeLimitIntegrationSuite) TestTerminateWorkflowCausedBySizeLimit() {
 	lastEvent := history.Events[len(history.Events)-1]
 	s.Equal(types.EventTypeWorkflowExecutionFailed, lastEvent.GetEventType())
 	failedEventAttributes := lastEvent.WorkflowExecutionFailedEventAttributes
-	s.Equal(common.FailureReasonSizeExceedsLimit, failedEventAttributes.GetReason())
+	s.Equal(common.FailureReasonHistorySizeExceedsLimit, failedEventAttributes.GetReason())
 
 	// verify visibility is correctly processed from open to close
 	isCloseCorrect := false

service/history/decision/checker.go

@@ -181,7 +181,7 @@ func (c *workflowSizeChecker) failWorkflowSizeExceedsLimit() (bool, error) {
 		)
 
 		attributes := &types.FailWorkflowExecutionDecisionAttributes{
-			Reason:  common.StringPtr(common.FailureReasonSizeExceedsLimit),
+			Reason:  common.StringPtr(common.FailureReasonHistorySizeExceedsLimit),
 			Details: []byte("Workflow history size / count exceeds limit."),
 		}
 

service/history/decision/checker_test.go

@@ -995,7 +995,7 @@ func TestWorkflowSizeChecker_failWorkflowSizeExceedsLimit(t *testing.T) {
 			}
 			if tc.expectFail {
 				mutableState.EXPECT().AddFailWorkflowEvent(testEventID, &types.FailWorkflowExecutionDecisionAttributes{
-					Reason:  common.StringPtr(common.FailureReasonSizeExceedsLimit),
+					Reason:  common.StringPtr(common.FailureReasonHistorySizeExceedsLimit),
 					Details: []byte("Workflow history size / count exceeds limit."),
 				}).Return(nil, nil).Times(1)
 			}

service/history/decision/task_handler_test.go

@@ -1606,7 +1606,7 @@ func TestHandleDecisions(t *testing.T) {
 		taskHandler.mutableState.(*execution.MockMutableState).EXPECT().GetNextEventID().Return(int64(12)) // nextEventID - 1 > historyCountLimit of 10
 		taskHandler.mutableState.(*execution.MockMutableState).EXPECT().GetExecutionInfo().Return(&persistence.WorkflowExecutionInfo{})
 		taskHandler.mutableState.(*execution.MockMutableState).EXPECT().AddFailWorkflowEvent(taskHandler.sizeLimitChecker.completedID, &types.FailWorkflowExecutionDecisionAttributes{
-			Reason:  common.StringPtr(common.FailureReasonSizeExceedsLimit),
+			Reason:  common.StringPtr(common.FailureReasonHistorySizeExceedsLimit),
 			Details: []byte("Workflow history size / count exceeds limit."),
 		}).Return(nil, errors.New("some error adding fail workflow event"))
 		res, err := taskHandler.handleDecisions(context.Background(), []byte{}, []*types.Decision{})

service/worker/diagnostics/invariant/failure/failure.go

@@ -57,6 +57,17 @@ func (f *failure) Check(ctx context.Context, params invariant.InvariantCheckInpu
 					Metadata:      invariant.MarshalData(FailureIssuesMetadata{Identity: identity}),
 				})
 				issueID++
+			} else if *reason == common.FailureReasonHistorySizeExceedsLimit {
+				result = append(result, invariant.InvariantCheckResult{
+					IssueID:       issueID,
+					InvariantType: WorkflowFailed.String(),
+					Reason:        HistorySizeExceedsLimit.String(),
+					Metadata: invariant.MarshalData(FailureIssuesMetadata{
+						Identity:      identity,
+						FailedEventID: event.ID,
+					}),
+				})
+				issueID++
 			} else {
 				result = append(result, invariant.InvariantCheckResult{
 					IssueID:       issueID,
@@ -171,6 +182,12 @@ func (f *failure) RootCause(ctx context.Context, params invariant.InvariantRootC
 				RootCause: invariant.RootCauseTypeServiceSideCustomError,
 				Metadata:  issue.Metadata,
 			})
+		case HistorySizeExceedsLimit.String():
+			result = append(result, invariant.InvariantRootCauseResult{
+				IssueID:   issue.IssueID,
+				RootCause: invariant.RootCauseTypeHistorySizeExceedsLimit,
+				Metadata:  issue.Metadata,
+			})
 		}
 	}
 	return result, nil

service/worker/diagnostics/invariant/failure/failure_test.go

@@ -35,7 +35,8 @@ import (
 )
 
 const (
-	testDomain = "test-domain"
+	testDomain                = "test-domain"
+	largeHistoryFailedEventID = int64(200001)
 )
 
 func Test__Check(t *testing.T) {
@@ -44,6 +45,12 @@ func Test__Check(t *testing.T) {
 	}
 	metadataInBytes, err := json.Marshal(metadata)
 	require.NoError(t, err)
+	largeHistoryMetadata := FailureIssuesMetadata{
+		Identity:      "localhost",
+		FailedEventID: largeHistoryFailedEventID,
+	}
+	largeHistoryMetadataInBytes, err := json.Marshal(largeHistoryMetadata)
+	require.NoError(t, err)
 	actMetadata := FailureIssuesMetadata{
 		Identity:            "localhost",
 		ActivityType:        "test-activity",
@@ -108,6 +115,19 @@ func Test__Check(t *testing.T) {
 			},
 			err: nil,
 		},
+		{
+			name:     "workflow history size limit exceeded",
+			testData: historySizeLimitExceededHistory(),
+			expectedResult: []invariant.InvariantCheckResult{
+				{
+					IssueID:       0,
+					InvariantType: WorkflowFailed.String(),
+					Reason:        HistorySizeExceedsLimit.String(),
+					Metadata:      largeHistoryMetadataInBytes,
+				},
+			},
+			err: nil,
+		},
 	}
 	for _, tc := range testCases {
 		inv := NewInvariant()
@@ -121,6 +141,29 @@ func Test__Check(t *testing.T) {
 	}
 }
 
+func historySizeLimitExceededHistory() *types.GetWorkflowExecutionHistoryResponse {
+	return &types.GetWorkflowExecutionHistoryResponse{
+		History: &types.History{
+			Events: []*types.HistoryEvent{
+				{
+					ID: 10,
+					DecisionTaskCompletedEventAttributes: &types.DecisionTaskCompletedEventAttributes{
+						Identity: "localhost",
+					},
+				},
+				{
+					ID: largeHistoryFailedEventID,
+					WorkflowExecutionFailedEventAttributes: &types.WorkflowExecutionFailedEventAttributes{
+						Reason:                       common.StringPtr(common.FailureReasonHistorySizeExceedsLimit),
+						Details:                      []byte("Workflow history size / count exceeds limit."),
+						DecisionTaskCompletedEventID: 10,
+					},
+				},
+			},
+		},
+	}
+}
+
 func failedWfHistory() *types.GetWorkflowExecutionHistoryResponse {
 	return &types.GetWorkflowExecutionHistoryResponse{
 		History: &types.History{
@@ -235,6 +278,11 @@ func Test__RootCause(t *testing.T) {
 	}
 	metadataInBytes, err := json.Marshal(metadata)
 	require.NoError(t, err)
+	largeHistoryMetadataInBytes, err := json.Marshal(FailureIssuesMetadata{
+		Identity:      "localhost",
+		FailedEventID: largeHistoryFailedEventID,
+	})
+	require.NoError(t, err)
 	testCases := []struct {
 		name           string
 		input          []invariant.InvariantCheckResult
@@ -289,6 +337,22 @@ func Test__RootCause(t *testing.T) {
 			}},
 			err: nil,
 		},
+		{
+			name: "workflow history size limit exceeded",
+			input: []invariant.InvariantCheckResult{
+				{
+					IssueID:       0,
+					InvariantType: WorkflowFailed.String(),
+					Reason:        HistorySizeExceedsLimit.String(),
+					Metadata:      largeHistoryMetadataInBytes,
+				}},
+			expectedResult: []invariant.InvariantRootCauseResult{{
+				IssueID:   0,
+				RootCause: invariant.RootCauseTypeHistorySizeExceedsLimit,
+				Metadata:  largeHistoryMetadataInBytes,
+			}},
+			err: nil,
+		},
 	}
 	inv := NewInvariant()
 	for _, tc := range testCases {

service/worker/diagnostics/invariant/failure/types.go

@@ -32,6 +32,7 @@ const (
 	HeartBeatBlobSizeLimit      ErrorType = "Heartbeat details has exceeded the blob size limit"
 	ActivityOutputBlobSizeLimit ErrorType = "Activity output has exceeded the blob size limit"
 	DecisionBlobSizeLimit       ErrorType = "Decision result caused to exceed blob size limit"
+	HistorySizeExceedsLimit     ErrorType = "The workflow history size or event count exceeded the configured limit"
 )
 
 func (e ErrorType) String() string {
@@ -55,6 +56,7 @@ type FailureIssuesMetadata struct {
 	ActivityType        string
 	ActivityScheduledID int64
 	ActivityStartedID   int64
+	FailedEventID       int64 `json:",omitempty"`
 }
 
 // BlobSizeMetadata includes the details of blob size limits

service/worker/diagnostics/invariant/interface.go

@@ -66,6 +66,7 @@ const (
 	RootCauseTypeServiceSidePanic                      RootCause = "There is a panic in the activity/workflow that is causing a failure"
 	RootCauseTypeServiceSideCustomError                RootCause = "Customised error returned by the activity/workflow"
 	RootCauseTypeBlobSizeLimit                         RootCause = "Workflow has exceeded the blob size limits configured for the domain"
+	RootCauseTypeHistorySizeExceedsLimit               RootCause = "Workflow history has exceeded the size or event count limit configured for the domain"
 )
 
 func (r RootCause) String() string {

service/worker/diagnostics/workflow.go

@@ -354,13 +354,16 @@ func retrieveFailureIssues(issues []invariant.InvariantCheckResult) ([]*failureI
 func retrieveFailureRootCause(rootCause []invariant.InvariantRootCauseResult) ([]*failureRootCauseResult, error) {
 	result := make([]*failureRootCauseResult, 0)
 	for _, rc := range rootCause {
-		if rc.RootCause == invariant.RootCauseTypeServiceSideIssue || rc.RootCause == invariant.RootCauseTypeServiceSidePanic || rc.RootCause == invariant.RootCauseTypeServiceSideCustomError {
+		switch rc.RootCause {
+		case invariant.RootCauseTypeServiceSideIssue,
+			invariant.RootCauseTypeServiceSidePanic,
+			invariant.RootCauseTypeServiceSideCustomError,
+			invariant.RootCauseTypeHistorySizeExceedsLimit:
 			result = append(result, &failureRootCauseResult{
 				IssueID:       rc.IssueID,
 				RootCauseType: rc.RootCause.String(),
 			})
-		}
-		if rc.RootCause == invariant.RootCauseTypeBlobSizeLimit {
+		case invariant.RootCauseTypeBlobSizeLimit:
 			var metadata failure.FailureRootcauseMetadata
 			err := json.Unmarshal(rc.Metadata, &metadata)
 			if err != nil {