Add limits on other workflow fields

common/dynamicconfig/constants.go

@@ -113,8 +113,18 @@ const (
 	MemoSizeLimitError = "limit.memoSize.error"
 	// MemoSizeLimitWarn is the per event memo size limit for warning
 	MemoSizeLimitWarn = "limit.memoSize.warn"
-	// NumPendingChildExecutionLimitError is the per workflow pending child workflow limit
-	NumPendingChildExecutionLimitError = "limit.numPendingChildExecution.error"
+	// NumPendingChildExecutionsLimitError is the maximum number of pending child workflows a workflow can have before
+	// StartChildWorkflowExecution commands will fail.
+	NumPendingChildExecutionsLimitError = "limit.numPendingChildExecutions.error"
+	// NumPendingActivitiesLimitError is the maximum number of pending activities a workflow can have before
+	// ScheduleActivityTask will fail.
+	NumPendingActivitiesLimitError = "limit.numPendingActivities.error"
+	// NumPendingSignalsLimitError is the maximum number of pending signals a workflow can have before
+	// SignalExternalWorkflowExecution commands from this workflow will fail.
+	NumPendingSignalsLimitError = "limit.numPendingSignals.error"
+	// NumPendingCancelRequestsLimitError is the maximum number of pending cancel requests a workflow can have before
+	// RequestCancelExternalWorkflowExecution commands will fail.
+	NumPendingCancelRequestsLimitError = "limit.numPendingCancelRequests.error"
 	// HistorySizeLimitError is the per workflow execution history size limit
 	HistorySizeLimitError = "limit.historySize.error"
 	// HistorySizeLimitWarn is the per workflow execution history size limit for warning

host/client_integration_test.go

@@ -81,11 +81,11 @@ func TestClientIntegrationSuite(t *testing.T) {
 }
 
 func (s *clientIntegrationSuite) SetupSuite() {
-	// maxPendingChildWorkflows should be NumPendingChildExecutionLimitError, but that value is 1000, and it takes
+	// maxPendingChildWorkflows should be NumPendingChildExecutionsLimit, but that value is 1000, and it takes
 	// too long to test (which is the point of this limit)
 	s.maxPendingChildExecutions = 10
 	s.dynamicConfigOverrides = map[dynamicconfig.Key]interface{}{
-		dynamicconfig.NumPendingChildExecutionLimitError: s.maxPendingChildExecutions,
+		dynamicconfig.NumPendingChildExecutionsLimitError: s.maxPendingChildExecutions,
 	}
 	s.setupSuite("testdata/clientintegrationtestcluster.yaml")
 

service/history/commandChecker.go

@@ -62,11 +62,14 @@ type (
 	}
 
 	workflowSizeLimits struct {
-		blobSizeLimitWarn                  int
-		blobSizeLimitError                 int
-		memoSizeLimitWarn                  int
-		memoSizeLimitError                 int
-		numPendingChildExecutionLimitError int
+		blobSizeLimitWarn              int
+		blobSizeLimitError             int
+		memoSizeLimitWarn              int
+		memoSizeLimitError             int
+		numPendingChildExecutionsLimit int
+		numPendingActivitiesLimit      int
+		numPendingSignalsLimit         int
+		numPendingCancelsRequestLimit  int
 	}
 
 	workflowSizeChecker struct {
@@ -189,7 +192,7 @@ func (c *workflowSizeChecker) checkIfNumChildWorkflowsExceedsLimit() error {
 	)
 
 	numPending := len(c.mutableState.GetPendingChildExecutionInfos())
-	errLimit := c.numPendingChildExecutionLimitError
+	errLimit := c.numPendingChildExecutionsLimit
 	if withinLimit(numPending, errLimit) {
 		return nil
 	}
@@ -199,7 +202,7 @@ func (c *workflowSizeChecker) checkIfNumChildWorkflowsExceedsLimit() error {
 			"has reached the error limit of %d established with %q",
 		numPending,
 		errLimit,
-		dynamicconfig.NumPendingChildExecutionLimitError,
+		dynamicconfig.NumPendingChildExecutionsLimitError,
 	)
 	logger.Error(err.Error(), tag.Error(err))
 	return err

service/history/commandChecker_test.go

@@ -718,7 +718,7 @@ func (s *commandAttrValidatorSuite) TestValidateCommandSequence_InvalidTerminalC
 func TestWorkflowSizeChecker_NumChildWorkflows(t *testing.T) {
 	const (
 		errMsg = "the number of pending child workflow executions, 1, " +
-			"has reached the error limit of 1 established with \"limit.numPendingChildExecution.error\""
+			"has reached the error limit of 1 established with \"limit.numPendingChildExecutions.error\""
 	)
 	for _, c := range []struct {
 		Name                      string
@@ -798,7 +798,7 @@ func TestWorkflowSizeChecker_NumChildWorkflows(t *testing.T) {
 			}
 
 			checker := newWorkflowSizeChecker(workflowSizeLimits{
-				numPendingChildExecutionLimitError: c.ErrorLimit,
+				numPendingChildExecutionsLimit: c.ErrorLimit,
 			}, mutableState, nil, nil, metricsHandler, logger)
 			err := checker.checkIfNumChildWorkflowsExceedsLimit()
 

service/history/configs/config.go

@@ -188,15 +188,18 @@ type Config struct {
 	DurableArchivalEnabled    dynamicconfig.BoolPropertyFn
 
 	// Size limit related settings
-	BlobSizeLimitError                 dynamicconfig.IntPropertyFnWithNamespaceFilter
-	BlobSizeLimitWarn                  dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MemoSizeLimitError                 dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MemoSizeLimitWarn                  dynamicconfig.IntPropertyFnWithNamespaceFilter
-	HistorySizeLimitError              dynamicconfig.IntPropertyFnWithNamespaceFilter
-	HistorySizeLimitWarn               dynamicconfig.IntPropertyFnWithNamespaceFilter
-	HistoryCountLimitError             dynamicconfig.IntPropertyFnWithNamespaceFilter
-	HistoryCountLimitWarn              dynamicconfig.IntPropertyFnWithNamespaceFilter
-	NumPendingChildExecutionLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
+	BlobSizeLimitError             dynamicconfig.IntPropertyFnWithNamespaceFilter
+	BlobSizeLimitWarn              dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MemoSizeLimitError             dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MemoSizeLimitWarn              dynamicconfig.IntPropertyFnWithNamespaceFilter
+	HistorySizeLimitError          dynamicconfig.IntPropertyFnWithNamespaceFilter
+	HistorySizeLimitWarn           dynamicconfig.IntPropertyFnWithNamespaceFilter
+	HistoryCountLimitError         dynamicconfig.IntPropertyFnWithNamespaceFilter
+	HistoryCountLimitWarn          dynamicconfig.IntPropertyFnWithNamespaceFilter
+	NumPendingChildExecutionsLimit dynamicconfig.IntPropertyFnWithNamespaceFilter
+	NumPendingActivitiesLimit      dynamicconfig.IntPropertyFnWithNamespaceFilter
+	NumPendingSignalsLimit         dynamicconfig.IntPropertyFnWithNamespaceFilter
+	NumPendingCancelsRequestLimit  dynamicconfig.IntPropertyFnWithNamespaceFilter
 
 	// DefaultActivityRetryOptions specifies the out-of-box retry policy if
 	// none is configured on the Activity by the user.
@@ -427,15 +430,18 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int32, isAdvancedVis
 			return false
 		},
 
-		BlobSizeLimitError:                 dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitError, 2*1024*1024),
-		BlobSizeLimitWarn:                  dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitWarn, 512*1024),
-		MemoSizeLimitError:                 dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitError, 2*1024*1024),
-		MemoSizeLimitWarn:                  dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitWarn, 2*1024),
-		NumPendingChildExecutionLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingChildExecutionLimitError, 1000),
-		HistorySizeLimitError:              dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitError, 50*1024*1024),
-		HistorySizeLimitWarn:               dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitWarn, 10*1024*1024),
-		HistoryCountLimitError:             dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitError, 50*1024),
-		HistoryCountLimitWarn:              dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitWarn, 10*1024),
+		BlobSizeLimitError:             dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitError, 2*1024*1024),
+		BlobSizeLimitWarn:              dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitWarn, 512*1024),
+		MemoSizeLimitError:             dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitError, 2*1024*1024),
+		MemoSizeLimitWarn:              dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitWarn, 2*1024),
+		NumPendingChildExecutionsLimit: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingChildExecutionsLimitError, 1000),
+		NumPendingActivitiesLimit:      dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingActivitiesLimitError, 1000),
+		NumPendingSignalsLimit:         dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingSignalsLimitError, 1000),
+		NumPendingCancelsRequestLimit:  dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingCancelRequestsLimitError, 1000),
+		HistorySizeLimitError:          dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitError, 50*1024*1024),
+		HistorySizeLimitWarn:           dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitWarn, 10*1024*1024),
+		HistoryCountLimitError:         dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitError, 50*1024),
+		HistoryCountLimitWarn:          dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitWarn, 10*1024),
 
 		ThrottledLogRPS:   dc.GetIntProperty(dynamicconfig.HistoryThrottledLogRPS, 4),
 		EnableStickyQuery: dc.GetBoolPropertyFnWithNamespaceFilter(dynamicconfig.EnableStickyQuery, true),

service/history/historyEngine2_test.go

@@ -991,7 +991,7 @@ func (s *engine2Suite) TestRespondWorkflowTaskCompleted_StartChildWorkflow_Excee
 	response := &persistence.GetWorkflowExecutionResponse{State: workflow.TestCloneToProto(ms)}
 	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(response, nil).AnyTimes()
 
-	s.historyEngine.shard.GetConfig().NumPendingChildExecutionLimitError = func(namespace string) int {
+	s.historyEngine.shard.GetConfig().NumPendingChildExecutionsLimit = func(namespace string) int {
 		return 5
 	}
 	_, err := s.historyEngine.RespondWorkflowTaskCompleted(metrics.AddMetricsContext(context.Background()), &historyservice.RespondWorkflowTaskCompletedRequest{

service/history/workflowTaskHandlerCallbacks.go

@@ -450,12 +450,14 @@ func (handler *workflowTaskHandlerCallbacksImpl) handleWorkflowTaskCompleted(
 		namespace := namespaceEntry.Name()
 		workflowSizeChecker := newWorkflowSizeChecker(
 			workflowSizeLimits{
-				blobSizeLimitWarn:  handler.config.BlobSizeLimitWarn(namespace.String()),
-				blobSizeLimitError: handler.config.BlobSizeLimitError(namespace.String()),
-				memoSizeLimitWarn:  handler.config.MemoSizeLimitWarn(namespace.String()),
-				memoSizeLimitError: handler.config.MemoSizeLimitError(namespace.String()),
-				numPendingChildExecutionLimitError: handler.config.NumPendingChildExecutionLimitError(namespace.
-					String()),
+				blobSizeLimitWarn:              handler.config.BlobSizeLimitWarn(namespace.String()),
+				blobSizeLimitError:             handler.config.BlobSizeLimitError(namespace.String()),
+				memoSizeLimitWarn:              handler.config.MemoSizeLimitWarn(namespace.String()),
+				memoSizeLimitError:             handler.config.MemoSizeLimitError(namespace.String()),
+				numPendingChildExecutionsLimit: handler.config.NumPendingChildExecutionsLimit(namespace.String()),
+				numPendingActivitiesLimit:      handler.config.NumPendingActivitiesLimit(namespace.String()),
+				numPendingSignalsLimit:         handler.config.NumPendingSignalsLimit(namespace.String()),
+				numPendingCancelsRequestLimit:  handler.config.NumPendingCancelsRequestLimit(namespace.String()),
 			},
 			ms,
 			handler.searchAttributesValidator,