Suggest continue-as-new before hitting maximumSignalsPerExecution

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,

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}
 }

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

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=

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

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
 }
 

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)
+	})
+}