diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 58af953efdc..a5b458fb5e8 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -2514,6 +2514,11 @@ system.transactionSizeLimit, since each batch is persisted within a single trans 10000, `MaximumSignalsPerExecution is max number of signals supported by single execution`, ) + MaximumSignalsPerExecutionSuggestContinueAsNewThreshold = NewNamespaceFloatSetting( + "history.maximumSignalsPerExecution.suggestContinueAsNewThreshold", + 0, + `MaximumSignalsPerExecutionSuggestContinueAsNewThreshold is the percentage threshold of total signals that any given workflow execution can receive before suggesting to continue-as-new. Set to zero to disable.`, + ) ShardUpdateMinInterval = NewGlobalDurationSetting( "history.shardUpdateMinInterval", 5*time.Minute, diff --git a/common/metrics/tags.go b/common/metrics/tags.go index afd8040b276..a35af1bb0fc 100644 --- a/common/metrics/tags.go +++ b/common/metrics/tags.go @@ -53,6 +53,7 @@ const ( suggestContinueAsNewReasonTooManyUpdates = "suggest_continue_as_new_reason_too_many_updates" suggestContinueAsNewReasonTooManyHistoryEvents = "suggest_continue_as_new_reason_too_many_history_events" suggestContinueAsNewReasonHistorySizeTooLarge = "suggest_continue_as_new_reason_history_size_too_large" + suggestContinueAsNewReasonTooManySignals = "suggest_continue_as_new_reason_too_many_signals" isFirstAttempt = "first-attempt" workflowStatus = "workflow_status" behaviorBefore = "behavior_before" @@ -508,6 +509,14 @@ func SuggestContinueAsNewReasonHistorySizeTooLargeTag(present bool) Tag { return Tag{Key: suggestContinueAsNewReasonHistorySizeTooLarge, Value: v} } +func SuggestContinueAsNewReasonTooManySignalsTag(present bool) Tag { + v := falseValue + if present { + v = trueValue + } + return Tag{Key: suggestContinueAsNewReasonTooManySignals, Value: v} +} + func WorkflowStatusTag(status string) Tag { return Tag{Key: workflowStatus, Value: status} } diff --git a/go.mod b/go.mod index b77311ca86a..184dacff87c 100644 --- a/go.mod +++ b/go.mod @@ -233,3 +233,7 @@ require ( sigs.k8s.io/structured-merge-diff/v6 v6.4.0 // indirect sigs.k8s.io/yaml v1.6.0 // indirect ) + +// TODO: remove this replace once temporalio/api#822 merges and a released +// go.temporal.io/api version contains SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_SIGNALS. +replace go.temporal.io/api => github.com/nitishagar/api-go v1.63.3-0.suggestsignals.1 diff --git a/go.sum b/go.sum index 8d0805ee2e2..ad84fb62a58 100644 --- a/go.sum +++ b/go.sum @@ -324,6 +324,8 @@ github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHv github.com/nexus-rpc/sdk-go v0.6.0 h1:QRgnP2zTbxEbiyWG/aXH8uSC5LV/Mg1fqb19jb4DBlo= github.com/nexus-rpc/sdk-go v0.6.0/go.mod h1:FHdPfVQwRuJFZFTF0Y2GOAxCrbIBNrcPna9slkGKPYk= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= +github.com/nitishagar/api-go v1.63.3-0.suggestsignals.1 h1:UrMHijY7y+NYNpSpOROxmthxw8uilXtmFl+cGdmiwus= +github.com/nitishagar/api-go v1.63.3-0.suggestsignals.1/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q= github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec= github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY= github.com/olivere/elastic/v7 v7.0.32 h1:R7CXvbu8Eq+WlsLgxmKVKPox0oOwAE/2T9Si5BnvK6E= @@ -471,8 +473,6 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= -go.temporal.io/api v1.63.2 h1:axAqAdHraBdREoK7Ufnl0LDvlmhDO0GmYJ1JTf495HA= -go.temporal.io/api v1.63.2/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q= go.temporal.io/auto-scaled-workers v0.0.0-20260706201056-4320b34799ee h1:y6A65Iml06cR3CpxW2Zn8FQLjniPDTnN4jtr69UZxXI= go.temporal.io/auto-scaled-workers v0.0.0-20260706201056-4320b34799ee/go.mod h1:hhHijO9XRPIkAflLJJHix61M9FzbRPqk8fSydkcLkqw= go.temporal.io/sdk v1.41.1 h1:yOpvsHyDD1lNuwlGBv/SUodCPhjv9nDeC9lLHW/fJUA= diff --git a/service/history/configs/config.go b/service/history/configs/config.go index 44e4de8dc07..7e02e7c03ef 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -194,10 +194,11 @@ type Config struct { ReplicatorProcessorMaxSkipTaskCount dynamicconfig.IntPropertyFn // System Limits - MaximumBufferedEventsBatch dynamicconfig.IntPropertyFn - MaximumBufferedEventsSizeInBytes dynamicconfig.IntPropertyFn - MaximumSignalsPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter - MaximumEventBatchSizeInBytes dynamicconfig.IntPropertyFn + MaximumBufferedEventsBatch dynamicconfig.IntPropertyFn + MaximumBufferedEventsSizeInBytes dynamicconfig.IntPropertyFn + MaximumSignalsPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter + MaximumSignalsPerExecutionSuggestContinueAsNewThreshold dynamicconfig.FloatPropertyFnWithNamespaceFilter + MaximumEventBatchSizeInBytes dynamicconfig.IntPropertyFn // ShardUpdateMinInterval is the minimum time interval within which the shard info can be updated. ShardUpdateMinInterval dynamicconfig.DurationPropertyFn @@ -635,15 +636,16 @@ func NewConfig( ReplicationStreamSenderLivenessMultiplier: dynamicconfig.ReplicationStreamSenderLivenessMultiplier.Get(dc), EnableHistoryReplicationRateLimiter: dynamicconfig.EnableHistoryReplicationRateLimiter.Get(dc), - MaximumBufferedEventsBatch: dynamicconfig.MaximumBufferedEventsBatch.Get(dc), - MaximumBufferedEventsSizeInBytes: dynamicconfig.MaximumBufferedEventsSizeInBytes.Get(dc), - MaximumSignalsPerExecution: dynamicconfig.MaximumSignalsPerExecution.Get(dc), - MaximumEventBatchSizeInBytes: dynamicconfig.MaximumEventBatchSizeInBytes.Get(dc), - ShardUpdateMinInterval: dynamicconfig.ShardUpdateMinInterval.Get(dc), - ShardFirstUpdateInterval: dynamicconfig.ShardFirstUpdateInterval.Get(dc), - ShardUpdateMinTasksCompleted: dynamicconfig.ShardUpdateMinTasksCompleted.Get(dc), - ShardSyncMinInterval: dynamicconfig.ShardSyncMinInterval.Get(dc), - ShardSyncTimerJitterCoefficient: dynamicconfig.TransferProcessorMaxPollIntervalJitterCoefficient.Get(dc), + MaximumBufferedEventsBatch: dynamicconfig.MaximumBufferedEventsBatch.Get(dc), + MaximumBufferedEventsSizeInBytes: dynamicconfig.MaximumBufferedEventsSizeInBytes.Get(dc), + MaximumSignalsPerExecution: dynamicconfig.MaximumSignalsPerExecution.Get(dc), + MaximumSignalsPerExecutionSuggestContinueAsNewThreshold: dynamicconfig.MaximumSignalsPerExecutionSuggestContinueAsNewThreshold.Get(dc), + MaximumEventBatchSizeInBytes: dynamicconfig.MaximumEventBatchSizeInBytes.Get(dc), + ShardUpdateMinInterval: dynamicconfig.ShardUpdateMinInterval.Get(dc), + ShardFirstUpdateInterval: dynamicconfig.ShardFirstUpdateInterval.Get(dc), + ShardUpdateMinTasksCompleted: dynamicconfig.ShardUpdateMinTasksCompleted.Get(dc), + ShardSyncMinInterval: dynamicconfig.ShardSyncMinInterval.Get(dc), + ShardSyncTimerJitterCoefficient: dynamicconfig.TransferProcessorMaxPollIntervalJitterCoefficient.Get(dc), // history client: client/history/client.go set the client timeout 30s // TODO: Return this value to the client: go.temporal.io/server/issues/294 diff --git a/service/history/workflow/workflow_task_state_machine.go b/service/history/workflow/workflow_task_state_machine.go index 85c6e529074..13b7bedce9e 100644 --- a/service/history/workflow/workflow_task_state_machine.go +++ b/service/history/workflow/workflow_task_state_machine.go @@ -557,6 +557,8 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskStartedEvent( slices.Contains(suggestContinueAsNewReasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_HISTORY_SIZE_TOO_LARGE)), metrics.SuggestContinueAsNewReasonTooManyHistoryEventsTag( slices.Contains(suggestContinueAsNewReasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_HISTORY_EVENTS)), + metrics.SuggestContinueAsNewReasonTooManySignalsTag( + slices.Contains(suggestContinueAsNewReasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_SIGNALS)), )).Record(1) } @@ -1511,6 +1513,14 @@ func (m *workflowTaskStateMachine) getHistorySizeInfo() (int64, []enumspb.Sugges if historyCount >= countLimit { reasons = append(reasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_HISTORY_EVENTS) } + maxSignals := config.MaximumSignalsPerExecution(namespaceName) + signalThreshold := config.MaximumSignalsPerExecutionSuggestContinueAsNewThreshold(namespaceName) + if maxSignals > 0 && signalThreshold > 0 { + signalSuggest := int(math.Ceil(float64(maxSignals) * signalThreshold)) + if signalSuggest > 0 && int(m.ms.GetExecutionInfo().SignalCount) >= signalSuggest { + reasons = append(reasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_SIGNALS) + } + } return historySize, reasons } diff --git a/service/history/workflow/workflow_task_state_machine_signals_test.go b/service/history/workflow/workflow_task_state_machine_signals_test.go new file mode 100644 index 00000000000..87075f26471 --- /dev/null +++ b/service/history/workflow/workflow_task_state_machine_signals_test.go @@ -0,0 +1,173 @@ +// The MIT License (MIT) +// +// Copyright (c) 2024 Temporal Technologies Inc. ALL RIGHTS RESERVED. +// +// See NOTICE.md for full restrictions. + +package workflow + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + enumspb "go.temporal.io/api/enums/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/common/cluster" + "go.temporal.io/server/components/callbacks" + "go.temporal.io/server/components/nexusoperations" + "go.temporal.io/server/service/history/events" + "go.temporal.io/server/service/history/hsm" + "go.temporal.io/server/service/history/shard" + "go.temporal.io/server/service/history/tests" + "go.uber.org/mock/gomock" +) + +// signalSuggestCANFixture builds a workflowTaskStateMachine whose only variable inputs are +// SignalCount and the two signal-related dynamic-config knobs, with the size/count suggest +// thresholds pushed out of the way so they cannot pollute the asserted reasons slice. +func signalSuggestCANFixture(t *testing.T, signalCount int64, maxSignals int, threshold float64) *workflowTaskStateMachine { + t.Helper() + controller := gomock.NewController(t) + config := tests.NewDynamicConfig() + // Push the size/count reasons far out so only the signal reason can fire. + config.HistorySizeSuggestContinueAsNew = func(string) int { return 1 << 60 } + config.HistoryCountSuggestContinueAsNew = func(string) int { return 1 << 60 } + config.MaximumSignalsPerExecution = func(string) int { return maxSignals } + config.MaximumSignalsPerExecutionSuggestContinueAsNewThreshold = func(string) float64 { return threshold } + + mockShard := shard.NewTestContext(controller, &persistencespb.ShardInfo{ShardId: 0, RangeId: 1}, config) + t.Cleanup(mockShard.StopForTest) + + reg := hsm.NewRegistry() + require.NoError(t, RegisterStateMachine(reg)) + require.NoError(t, callbacks.RegisterStateMachine(reg)) + require.NoError(t, nexusoperations.RegisterStateMachines(reg)) + mockShard.SetStateMachineRegistry(reg) + + namespaceEntry := tests.GlobalNamespaceEntry + mockShard.Resource.NamespaceCache.EXPECT().GetNamespaceByID(tests.NamespaceID).Return(namespaceEntry, nil).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().ClusterNameForFailoverVersion(namespaceEntry.IsGlobalNamespace(), namespaceEntry.FailoverVersion(tests.WorkflowID)).Return(cluster.TestCurrentClusterName).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().GetClusterID().Return(int64(1)).AnyTimes() + + ms := NewMutableState(mockShard, events.NewMockCache(controller), mockShard.GetLogger(), namespaceEntry, tests.WorkflowID, tests.RunID, time.Now().UTC()) + // getHistorySizeInfo returns early when ExecutionStats is nil, so set a non-nil one. + ms.GetExecutionInfo().ExecutionStats = &persistencespb.ExecutionStats{} + ms.GetExecutionInfo().SignalCount = signalCount + return newWorkflowTaskStateMachine(ms, mockShard.GetMetricsHandler()) +} + +func TestGetHistorySizeInfo_SignalSuggestCAN(t *testing.T) { + const signalReason = enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_SIGNALS + // With max=10, threshold=0.5 the boundary is ceil(10*0.5)=5: the reason fires at SignalCount>=5. + const maxSignals = 10 + const threshold = 0.5 + + t.Run("disabled by zero threshold", func(t *testing.T) { + // INV-3: threshold 0 -> never suggest, regardless of count. + m := signalSuggestCANFixture(t, 100, maxSignals, 0) + _, reasons := m.getHistorySizeInfo() + require.NotContains(t, reasons, signalReason, "threshold 0 must disable the signal reason") + require.Empty(t, reasons, "no reason should fire when only the (disabled) signal threshold is configured") + }) + + t.Run("disabled by zero max signals", func(t *testing.T) { + // INV-3: maxSignals 0 (hard limit disabled) -> signal suggestion also disabled. + m := signalSuggestCANFixture(t, 100, 0, threshold) + _, reasons := m.getHistorySizeInfo() + require.NotContains(t, reasons, signalReason, "maxSignals 0 must disable the signal reason") + require.Empty(t, reasons) + }) + + t.Run("below boundary does not suggest", func(t *testing.T) { + // boundary=5; count=4 is one short -> no reason. + m := signalSuggestCANFixture(t, 4, maxSignals, threshold) + _, reasons := m.getHistorySizeInfo() + require.NotContains(t, reasons, signalReason) + require.Empty(t, reasons) + }) + + t.Run("at boundary suggests (inclusive)", func(t *testing.T) { + // INV-3: inclusive >= at exactly ceil(max*threshold). + m := signalSuggestCANFixture(t, 5, maxSignals, threshold) + _, reasons := m.getHistorySizeInfo() + require.Contains(t, reasons, signalReason) + }) + + t.Run("above boundary suggests", func(t *testing.T) { + m := signalSuggestCANFixture(t, 9, maxSignals, threshold) + _, reasons := m.getHistorySizeInfo() + require.Contains(t, reasons, signalReason) + }) + + t.Run("threshold 1.0 suggests at hard limit", func(t *testing.T) { + // INV-3/INV-2: threshold 1.0 -> suggest as soon as count >= max (the hard-limit instant). + m := signalSuggestCANFixture(t, int64(maxSignals), maxSignals, 1.0) + _, reasons := m.getHistorySizeInfo() + require.Contains(t, reasons, signalReason) + }) + + t.Run("threshold 1.0 below hard limit does not suggest", func(t *testing.T) { + m := signalSuggestCANFixture(t, int64(maxSignals)-1, maxSignals, 1.0) + _, reasons := m.getHistorySizeInfo() + require.NotContains(t, reasons, signalReason) + require.Empty(t, reasons) + }) + + t.Run("coexists with history-size reason", func(t *testing.T) { + // INV-7: when the history-size threshold AND the signal threshold are both crossed, both + // reasons must appear in the slice (additive, not mutually exclusive). The shared fixture + // pushes the size/count limits out of the way, so build a dedicated state machine here that + // sets a low size threshold alongside a crossed signal threshold. + controller := gomock.NewController(t) + config := tests.NewDynamicConfig() + config.HistorySizeSuggestContinueAsNew = func(string) int { return 1 } // tiny: any history fires it + config.HistoryCountSuggestContinueAsNew = func(string) int { return 1 << 60 } + config.MaximumSignalsPerExecution = func(string) int { return maxSignals } + config.MaximumSignalsPerExecutionSuggestContinueAsNewThreshold = func(string) float64 { return threshold } + + mockShard := shard.NewTestContext(controller, &persistencespb.ShardInfo{ShardId: 0, RangeId: 1}, config) + t.Cleanup(mockShard.StopForTest) + + reg := hsm.NewRegistry() + require.NoError(t, RegisterStateMachine(reg)) + require.NoError(t, callbacks.RegisterStateMachine(reg)) + require.NoError(t, nexusoperations.RegisterStateMachines(reg)) + mockShard.SetStateMachineRegistry(reg) + + namespaceEntry := tests.GlobalNamespaceEntry + mockShard.Resource.NamespaceCache.EXPECT().GetNamespaceByID(tests.NamespaceID).Return(namespaceEntry, nil).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().ClusterNameForFailoverVersion(namespaceEntry.IsGlobalNamespace(), namespaceEntry.FailoverVersion(tests.WorkflowID)).Return(cluster.TestCurrentClusterName).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + mockShard.Resource.ClusterMetadata.EXPECT().GetClusterID().Return(int64(1)).AnyTimes() + + ms := NewMutableState(mockShard, events.NewMockCache(controller), mockShard.GetLogger(), namespaceEntry, tests.WorkflowID, tests.RunID, time.Now().UTC()) + ms.GetExecutionInfo().ExecutionStats = &persistencespb.ExecutionStats{HistorySize: 1 << 20} + // SignalCount above the boundary (ceil(10*0.5)=5). + ms.GetExecutionInfo().SignalCount = int64(maxSignals) + m := newWorkflowTaskStateMachine(ms, mockShard.GetMetricsHandler()) + + _, reasons := m.getHistorySizeInfo() + require.Contains(t, reasons, signalReason, "signal reason must fire when its threshold is crossed") + require.Contains(t, reasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_HISTORY_SIZE_TOO_LARGE, + "history-size reason must coexist when its threshold is also crossed") + require.Len(t, reasons, 2, "exactly the two crossed reasons, additive") + }) + + t.Run("new run after continue-as-new does not inherit prior signal count", func(t *testing.T) { + // INV-4: SignalCount is per-execution and resets on continue-as-new (a new run gets a new + // mutable state). A run whose count crossed the threshold suggests; the successor run, with + // a fresh low SignalCount, must NOT suggest even though the prior run did. Each call to + // signalSuggestCANFixture builds an independent mutable state, modelling the two runs. + priorRun := signalSuggestCANFixture(t, int64(maxSignals), maxSignals, threshold) + _, priorReasons := priorRun.getHistorySizeInfo() + require.Contains(t, priorReasons, signalReason, "prior run over the threshold must suggest") + + newRun := signalSuggestCANFixture(t, 0, maxSignals, threshold) + _, newReasons := newRun.getHistorySizeInfo() + require.NotContains(t, newReasons, signalReason, + "new run's fresh SignalCount must not trigger the reason even though the prior run suggested") + require.Empty(t, newReasons) + }) +} diff --git a/tests/signal_suggest_can_test.go b/tests/signal_suggest_can_test.go new file mode 100644 index 00000000000..fa028fbdd7e --- /dev/null +++ b/tests/signal_suggest_can_test.go @@ -0,0 +1,240 @@ +// The MIT License (MIT) +// +// Copyright (c) 2024 Temporal Technologies Inc. ALL RIGHTS RESERVED. +// +// See NOTICE.md for full restrictions. + +package tests + +import ( + "testing" + "time" + + "github.com/google/uuid" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/payloads" + "go.temporal.io/server/common/testing/parallelsuite" + "go.temporal.io/server/service/history/consts" + "go.temporal.io/server/tests/testcore" + "google.golang.org/protobuf/types/known/durationpb" +) + +type SignalSuggestCANTestSuite struct { + parallelsuite.Suite[*SignalSuggestCANTestSuite] +} + +func TestSignalSuggestCANTestSuiteLegacy(t *testing.T) { + parallelsuite.Run(t, &SignalSuggestCANTestSuite{}, []testcore.TestOption{}) +} + +// maxSignals and threshold below are chosen so the suggest boundary is ceil(5*0.6)=3: a third +// signal flips SuggestContinueAsNew on with the TOO_MANY_SIGNALS reason, while the hard limit +// (5) still rejects the sixth signal. Both knobs are namespace-scoped, so NewEnv applies them as +// a namespace constraint on the shared cluster (no dedicated cluster required). +const ( + signalSuggestMaxSignals = 5 + signalSuggestThreshold = 0.6 + signalSuggestBoundary = 3 // ceil(5 * 0.6) + signalSuggestSignalName = "suggest-can-signal" + signalSuggestWorkflow = "signal-suggest-can-type" + signalSuggestTaskQueue = "signal-suggest-can-taskqueue" +) + +func (s *SignalSuggestCANTestSuite) TestSignalSuggestCAN(opts []testcore.TestOption) { + env := testcore.NewEnv(s.T(), append(opts, + testcore.WithDynamicConfig(dynamicconfig.MaximumSignalsPerExecution, signalSuggestMaxSignals), + testcore.WithDynamicConfig(dynamicconfig.MaximumSignalsPerExecutionSuggestContinueAsNewThreshold, signalSuggestThreshold), + )...) + identity := "worker1" + id := "functional-signal-suggest-can-test-" + uuid.NewString() + workflowExecution := &commonpb.WorkflowExecution{WorkflowId: id} + taskQueue := &taskqueuepb.TaskQueue{Name: signalSuggestTaskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + + // The workflow stays open (completes only on a "finish" signal) so we can send signals and + // observe successive WorkflowTaskStarted attributes as SignalCount climbs. + startResp, err := env.FrontendClient().StartWorkflowExecution(env.Context(), &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: env.Namespace().String(), + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: signalSuggestWorkflow}, + TaskQueue: taskQueue, + WorkflowRunTimeout: durationpb.New(100 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: identity, + }) + s.NoError(err) + workflowExecution.RunId = startResp.RunId + + suggestCANSeen := false + signalReasonSeen := false + wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) { + for _, event := range task.History.Events { + if event.GetEventType() != enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED { + continue + } + attrs := event.GetWorkflowTaskStartedEventAttributes() + if attrs.GetSuggestContinueAsNew() { + suggestCANSeen = true + for _, reason := range attrs.GetSuggestContinueAsNewReasons() { + if reason == enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_SIGNALS { + signalReasonSeen = true + } + } + } + } + // Complete the workflow once the finish signal has been received. + for _, event := range task.History.Events { + if event.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED && + event.GetWorkflowExecutionSignaledEventAttributes().GetSignalName() == "finish" { + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{ + Result: payloads.EncodeString("Done"), + }}, + }}, nil + } + } + return []*commandpb.Command{}, nil + } + //nolint:staticcheck // SA1019 TaskPoller replacement needed + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: env.Namespace().String(), + TaskQueue: taskQueue, + Identity: identity, + WorkflowTaskHandler: wtHandler, + Logger: env.Logger, + T: s.T(), + } + + // Drain the initial workflow task. + _, err = poller.PollAndProcessWorkflowTask() + s.NoError(err) + + // Send signals up to the suggest boundary (ceil(max*threshold)=3). Each signal below the hard + // limit (5) must be accepted and must produce a workflow task whose started event we inspect. + for i := 0; i < signalSuggestBoundary; i++ { + _, err = env.FrontendClient().SignalWorkflowExecution(env.Context(), &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: env.Namespace().String(), + WorkflowExecution: workflowExecution, + SignalName: signalSuggestSignalName, + Input: payloads.EncodeString("signal"), + Identity: identity, + }) + s.NoError(err) + _, err = poller.PollAndProcessWorkflowTask() + s.NoError(err) + } + // After `boundary` signals the suggestion should have fired. + s.True(suggestCANSeen, "expected SuggestContinueAsNew=true once SignalCount reached the boundary") + s.True(signalReasonSeen, "expected SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_SIGNALS in reasons") + + // INV-2: the hard limit still rejects beyond maximumSignalsPerExecution. The three signals + // above are under max (5), so the 4th and 5th are still accepted; the 6th crosses the limit. + for i := signalSuggestBoundary; i < signalSuggestMaxSignals; i++ { + _, err = env.FrontendClient().SignalWorkflowExecution(env.Context(), &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: env.Namespace().String(), + WorkflowExecution: workflowExecution, + SignalName: signalSuggestSignalName, + Input: payloads.EncodeString("signal"), + Identity: identity, + }) + s.NoError(err) + } + _, err = env.FrontendClient().SignalWorkflowExecution(env.Context(), &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: env.Namespace().String(), + WorkflowExecution: workflowExecution, + SignalName: signalSuggestSignalName, + Input: payloads.EncodeString("signal"), + Identity: identity, + }) + s.ErrorIs(err, consts.ErrSignalsLimitExceeded, "hard limit must still reject beyond maximumSignalsPerExecution") + + // Complete the workflow so the test can tear down cleanly. + _, err = env.FrontendClient().SignalWorkflowExecution(env.Context(), &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: env.Namespace().String(), + WorkflowExecution: workflowExecution, + SignalName: "finish", + Input: payloads.EncodeString("finish"), + Identity: identity, + }) + s.NoError(err) + _, err = poller.PollAndProcessWorkflowTask() + s.NoError(err) +} + +func (s *SignalSuggestCANTestSuite) TestSignalSuggestCAN_DisabledByDefault(opts []testcore.TestOption) { + // Backward compatibility: with the default threshold (0) the signal reason must never appear, + // even with a tiny hard limit. + env := testcore.NewEnv(s.T(), append(opts, + testcore.WithDynamicConfig(dynamicconfig.MaximumSignalsPerExecution, signalSuggestMaxSignals), + )...) + identity := "worker1" + id := "functional-signal-suggest-can-disabled-" + uuid.NewString() + workflowExecution := &commonpb.WorkflowExecution{WorkflowId: id} + taskQueue := &taskqueuepb.TaskQueue{Name: signalSuggestTaskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + + startResp, err := env.FrontendClient().StartWorkflowExecution(env.Context(), &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: env.Namespace().String(), + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: signalSuggestWorkflow}, + TaskQueue: taskQueue, + WorkflowRunTimeout: durationpb.New(100 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: identity, + }) + s.NoError(err) + workflowExecution.RunId = startResp.RunId + + signalReasonSeen := false + wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) { + for _, event := range task.History.Events { + if event.GetEventType() != enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED { + continue + } + for _, reason := range event.GetWorkflowTaskStartedEventAttributes().GetSuggestContinueAsNewReasons() { + if reason == enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_SIGNALS { + signalReasonSeen = true + } + } + } + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{ + Result: payloads.EncodeString("Done"), + }}, + }}, nil + } + //nolint:staticcheck // SA1019 TaskPoller replacement needed + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: env.Namespace().String(), + TaskQueue: taskQueue, + Identity: identity, + WorkflowTaskHandler: wtHandler, + Logger: env.Logger, + T: s.T(), + } + _, err = poller.PollAndProcessWorkflowTask() + s.NoError(err) + + // Send one signal (below the hard limit) and process the task. + _, err = env.FrontendClient().SignalWorkflowExecution(env.Context(), &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: env.Namespace().String(), + WorkflowExecution: workflowExecution, + SignalName: signalSuggestSignalName, + Input: payloads.EncodeString("signal"), + Identity: identity, + }) + s.NoError(err) + _, err = poller.PollAndProcessWorkflowTask() + s.NoError(err) + + s.False(signalReasonSeen, "signal reason must not appear when threshold is 0 (disabled default)") +}