diff --git a/service/frontend/namespace_handler.go b/service/frontend/namespace_handler.go index 78f8a29a62..06262c92a9 100644 --- a/service/frontend/namespace_handler.go +++ b/service/frontend/namespace_handler.go @@ -916,8 +916,9 @@ func (d *namespaceHandler) createResponse( WorkflowTaskCompletionPagination: d.config.EnableWorkflowTaskCompletionPagination(info.Name), }, Limits: &namespacepb.NamespaceInfo_Limits{ - BlobSizeLimitError: int64(d.config.BlobSizeLimitError(info.Name)), - MemoSizeLimitError: int64(d.config.MemoSizeLimitError(info.Name)), + BlobSizeLimitError: int64(d.config.BlobSizeLimitError(info.Name)), + MemoSizeLimitError: int64(d.config.MemoSizeLimitError(info.Name)), + WorkflowTaskCompletionSizeLimitError: int64(d.config.WorkflowTaskCompletionBufferSizeLimit(info.Name)), }, SupportsSchedules: d.config.EnableSchedules(info.Name), } diff --git a/service/frontend/namespace_handler_test.go b/service/frontend/namespace_handler_test.go index 64aebfa4cb..043942f821 100644 --- a/service/frontend/namespace_handler_test.go +++ b/service/frontend/namespace_handler_test.go @@ -398,6 +398,7 @@ func (s *namespaceHandlerCommonSuite) TestCapabilitiesAndLimits() { s.False(resp.NamespaceInfo.Capabilities.WorkflowTaskCompletionPagination) s.Equal(int64(2*1024*1024), resp.NamespaceInfo.Limits.BlobSizeLimitError) s.Equal(int64(2*1024*1024), resp.NamespaceInfo.Limits.MemoSizeLimitError) + s.Equal(int64(40*1024*1024), resp.NamespaceInfo.Limits.WorkflowTaskCompletionSizeLimitError) // Second call: Override the default value of dynamic configs. s.config.EnableEagerWorkflowStart = dc.GetBoolPropertyFnFilteredByNamespace(false) @@ -415,6 +416,7 @@ func (s *namespaceHandlerCommonSuite) TestCapabilitiesAndLimits() { s.config.WorkerCommandsEnabled = dc.GetBoolPropertyFnFilteredByNamespace(true) s.config.PollerAutoscalingAutoEnroll = dc.GetBoolPropertyFnFilteredByNamespace(true) s.config.EnableWorkflowTaskCompletionPagination = dc.GetBoolPropertyFnFilteredByNamespace(true) + s.config.WorkflowTaskCompletionBufferSizeLimit = dc.GetIntPropertyFnFilteredByNamespace(4096) resp, err = s.handler.DescribeNamespace(context.Background(), &workflowservice.DescribeNamespaceRequest{ Namespace: "ns", @@ -434,6 +436,7 @@ func (s *namespaceHandlerCommonSuite) TestCapabilitiesAndLimits() { s.True(resp.NamespaceInfo.Capabilities.WorkflowTaskCompletionPagination) s.Equal(int64(1024), resp.NamespaceInfo.Limits.BlobSizeLimitError) s.Equal(int64(512), resp.NamespaceInfo.Limits.MemoSizeLimitError) + s.Equal(int64(4096), resp.NamespaceInfo.Limits.WorkflowTaskCompletionSizeLimitError) s.config.Activity.StartDelayEnabled = dc.GetBoolPropertyFnFilteredByNamespace(true) s.config.Activity.EnableStandaloneActivityOperatorCommands = dc.GetBoolPropertyFnFilteredByNamespace(true) diff --git a/service/frontend/service.go b/service/frontend/service.go index 542406c155..4383464d7a 100644 --- a/service/frontend/service.go +++ b/service/frontend/service.go @@ -242,6 +242,7 @@ type Config struct { WorkflowTimeSkippingMaxSkipPerSession dynamicconfig.IntPropertyFnWithNamespaceFilter StandaloneNexusOperationsEnabled dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableWorkflowTaskCompletionPagination dynamicconfig.BoolPropertyFnWithNamespaceFilter + WorkflowTaskCompletionBufferSizeLimit dynamicconfig.IntPropertyFnWithNamespaceFilter HTTPAllowedHosts dynamicconfig.TypedPropertyFn[*regexp.Regexp] AllowedExperiments dynamicconfig.TypedPropertyFnWithNamespaceFilter[[]string] @@ -420,6 +421,7 @@ func NewConfig( WorkflowTimeSkippingMaxSkipPerSession: dynamicconfig.WorkflowTimeSkippingMaxSkipPerSession.Get(dc), StandaloneNexusOperationsEnabled: chasmnexus.Enabled.Get(dc), EnableWorkflowTaskCompletionPagination: dynamicconfig.EnableWorkflowTaskCompletionPagination.Get(dc), + WorkflowTaskCompletionBufferSizeLimit: dynamicconfig.WorkflowTaskCompletionBufferSizeLimit.Get(dc), HTTPAllowedHosts: dynamicconfig.FrontendHTTPAllowedHosts.Get(dc), AllowedExperiments: dynamicconfig.FrontendAllowedExperiments.Get(dc), diff --git a/service/history/api/respondworkflowtaskcompleted/api.go b/service/history/api/respondworkflowtaskcompleted/api.go index 58eb2888f1..3a6e1f551a 100644 --- a/service/history/api/respondworkflowtaskcompleted/api.go +++ b/service/history/api/respondworkflowtaskcompleted/api.go @@ -383,7 +383,7 @@ func (handler *WorkflowTaskCompletedHandler) Invoke( if paginationOverflow { // Per-workflow completion buffer overflowed: terminate the workflow wtFailedCause = newWorkflowTaskFailedCause( - enumspb.WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_REQUEST_TOO_LARGE, serviceerror.NewInvalidArgument( "workflow task completion buffer size exceeds the per-workflow limit"), true) diff --git a/tests/workflow_completion_pagination_test.go b/tests/workflow_completion_pagination_test.go index 401dee0574..429a96efdc 100644 --- a/tests/workflow_completion_pagination_test.go +++ b/tests/workflow_completion_pagination_test.go @@ -313,6 +313,11 @@ func (s *WorkflowCompletionPaginationTestSuite) TestBufferOverflowFailsWorkflowT history := env.GetHistory(env.Namespace().String(), we) s.Equal(1, countEvents(history, enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED)) s.Equal(0, countEvents(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED)) + for _, e := range history { + if e.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED { + s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_REQUEST_TOO_LARGE, e.GetWorkflowTaskFailedEventAttributes().GetCause()) + } + } } // TestOutOfOrderPagesReassemble verifies that pages buffered out of arrival order