Implement workflow ID reuse rate limtier

common/dynamicconfig/constants.go

@@ -2756,6 +2756,18 @@ that task will be sent to DLQ.`,
 		false,
 		`If true, validate the start time of the old workflow is older than WorkflowIdReuseMinimalInterval when reusing workflow ID.`,
 	)
+	WorkflowIDReuseRate = NewNamespaceIntSetting(
+		"history.workflowIDReuseRate",
+		0,
+		`WorkflowIDReuseRate limits the rate of new workflow execution creation per
+(namespace, workflowID) pair on a single history host. 0 = disabled (default).`,
+	)
+	WorkflowIDReuseBurstRatio = NewNamespaceFloatSetting(
+		"history.workflowIDReuseBurstRatio",
+		1.0,
+		`WorkflowIDReuseBurstRatio is the burst-to-rate ratio for the per-(namespace, workflowID)
+start rate limiter. Burst = max(1, int(rps * ratio)). Default 1.0 (no burst above rate).`,
+	)
 	HealthPersistenceLatencyFailure = NewGlobalFloatSetting(
 		"history.healthPersistenceLatencyFailure",
 		500,

service/history/api/resetworkflow/api.go

@@ -18,6 +18,7 @@ import (
 	"go.temporal.io/server/common/persistence/versionhistory"
 	"go.temporal.io/server/common/worker_versioning"
 	"go.temporal.io/server/service/history/api"
+	"go.temporal.io/server/service/history/consts"
 	historyi "go.temporal.io/server/service/history/interfaces"
 	"go.temporal.io/server/service/history/ndc"
 )
@@ -40,6 +41,10 @@ func Invoke(
 
 	request := resetRequest.ResetRequest
 	workflowID := request.WorkflowExecution.GetWorkflowId()
+
+	if rl := shardContext.WorkflowIDReuseRL(namespaceID, workflowID); rl != nil && !rl.Allow() {
+		return nil, consts.ErrWorkflowIDRateLimitExceeded
+	}
 	baseRunID := request.WorkflowExecution.GetRunId()
 
 	baseWorkflowLease, err := workflowConsistencyChecker.GetWorkflowLease(

service/history/api/signalwithstartworkflow/signal_with_start_workflow.go

@@ -256,6 +256,7 @@ func startAndSignalWithoutCurrentWorkflow(
 		newWorkflowLease.GetMutableState(),
 		newWorkflow,
 		newWorkflowEventsSeq,
+		historyi.TransactionPolicyActive,
 	)
 	switch failedErr := err.(type) {
 	case nil:

service/history/api/startworkflow/api.go

@@ -319,6 +319,7 @@ func (s *Starter) createBrandNew(ctx context.Context, creationParams *creationPa
 		creationParams.workflowLease.GetMutableState(),
 		creationParams.workflowSnapshot,
 		creationParams.workflowEventBatches,
+		historyi.TransactionPolicyActive,
 	)
 }
 
@@ -394,6 +395,7 @@ func (s *Starter) createAsCurrent(
 		creationParams.workflowLease.GetMutableState(),
 		creationParams.workflowSnapshot,
 		creationParams.workflowEventBatches,
+		historyi.TransactionPolicyActive,
 	)
 }
 

service/history/chasm_engine.go

@@ -862,6 +862,7 @@ func (e *ChasmEngine) persistAsBrandNew(
 		newExecutionParams.mutableState,
 		newExecutionParams.snapshot,
 		newExecutionParams.events,
+		historyi.TransactionPolicyActive,
 	)
 	if err == nil {
 		return currentExecutionInfo{}, false, nil
@@ -1047,6 +1048,7 @@ func (e *ChasmEngine) handleReusePolicy(
 		newExecutionParams.mutableState,
 		newExecutionParams.snapshot,
 		newExecutionParams.events,
+		historyi.TransactionPolicyActive,
 	)
 	if err != nil {
 		return chasm.StartExecutionResult{}, err

service/history/configs/config.go

@@ -395,6 +395,8 @@ type Config struct {
 
 	WorkflowIdReuseMinimalInterval           dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	EnableWorkflowIdReuseStartTimeValidation dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	WorkflowIDReuseRate                      dynamicconfig.IntPropertyFnWithNamespaceFilter
+	WorkflowIDReuseBurstRatio                dynamicconfig.FloatPropertyFnWithNamespaceFilter
 
 	HealthPersistenceLatencyFailure dynamicconfig.FloatPropertyFn
 	HealthPersistenceErrorRatio     dynamicconfig.FloatPropertyFn
@@ -771,6 +773,8 @@ func NewConfig(
 		SendRawWorkflowHistory:                   dynamicconfig.SendRawWorkflowHistory.Get(dc),
 		WorkflowIdReuseMinimalInterval:           dynamicconfig.WorkflowIdReuseMinimalInterval.Get(dc),
 		EnableWorkflowIdReuseStartTimeValidation: dynamicconfig.EnableWorkflowIdReuseStartTimeValidation.Get(dc),
+		WorkflowIDReuseRate:            dynamicconfig.WorkflowIDReuseRate.Get(dc),
+		WorkflowIDReuseBurstRatio:                dynamicconfig.WorkflowIDReuseBurstRatio.Get(dc),
 
 		HealthPersistenceLatencyFailure: dynamicconfig.HealthPersistenceLatencyFailure.Get(dc),
 		HealthPersistenceErrorRatio:     dynamicconfig.HealthPersistenceErrorRatio.Get(dc),

service/history/consts/const.go

@@ -137,6 +137,12 @@ var (
 
 	// ErrResetRedirectLimitReached indicates a possible long chain (or a loop) of resets that cannot be handled.
 	ErrResetRedirectLimitReached = serviceerror.NewInternal("The chain of resets is too long to iterate.")
+	// ErrWorkflowIDRateLimitExceeded is returned when the per-(namespace, workflowID) start rate limit is exceeded.
+	ErrWorkflowIDRateLimitExceeded = &serviceerror.ResourceExhausted{
+		Cause:   enumspb.RESOURCE_EXHAUSTED_CAUSE_RPS_LIMIT,
+		Scope:   enumspb.RESOURCE_EXHAUSTED_SCOPE_NAMESPACE,
+		Message: "workflow ID start rate limit exceeded",
+	}
 )
 
 // StaleStateError is an indicator that after loading the state for a task it was detected as stale. It's possible that

service/history/interfaces/shard_context.go

@@ -21,6 +21,7 @@ import (
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/serialization"
+	"go.temporal.io/server/common/quotas"
 	"go.temporal.io/server/common/pingable"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/service/history/configs"
@@ -106,6 +107,8 @@ type (
 		GetFinalizer() *finalizer.Finalizer
 
 		ChasmRegistry() *chasm.Registry
+
+		WorkflowIDReuseRL(namespaceID namespace.ID, workflowID string) quotas.RateLimiter
 	}
 
 	// A ControllableContext is a Context plus other methods needed by

service/history/interfaces/workflow_context.go

@@ -54,6 +54,7 @@ type (
 			newMutableState MutableState,
 			newWorkflow *persistence.WorkflowSnapshot,
 			newWorkflowEvents []*persistence.WorkflowEvents,
+			transactionPolicy TransactionPolicy,
 		) error
 		ConflictResolveWorkflowExecution(
 			ctx context.Context,

service/history/ndc/transaction_manager_new_workflow.go

@@ -164,6 +164,7 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) createAsCurrent(
 			targetWorkflow.GetMutableState(),
 			targetWorkflowSnapshot,
 			targetWorkflowEventsSeq,
+			historyi.TransactionPolicyPassive,
 		)
 	}
 
@@ -180,6 +181,7 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) createAsCurrent(
 		targetWorkflow.GetMutableState(),
 		targetWorkflowSnapshot,
 		targetWorkflowEventsSeq,
+		historyi.TransactionPolicyPassive,
 	)
 }
 
@@ -254,6 +256,7 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) createAsZombie(
 		ms,
 		targetWorkflowSnapshot,
 		targetWorkflowEventsSeq,
+		historyi.TransactionPolicyPassive,
 	)
 	switch err.(type) {
 	case nil:

service/history/ndc/transaction_manager_new_workflow_test.go

@@ -122,6 +122,7 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_BrandNew(
 		mutableState,
 		workflowSnapshot,
 		workflowEventsSeq,
+		gomock.Any(),
 	).Return(nil)
 
 	err := s.createMgr.dispatchForNewWorkflow(ctx, chasm.WorkflowArchetypeID, newWorkflow)
@@ -195,6 +196,7 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_CreateAsC
 		targetMutableState,
 		targetWorkflowSnapshot,
 		targetWorkflowEventsSeq,
+		gomock.Any(),
 	).Return(nil)
 
 	err := s.createMgr.dispatchForNewWorkflow(ctx, chasm.WorkflowArchetypeID, targetWorkflow)
@@ -264,6 +266,7 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_CreateAsZ
 		targetMutableState,
 		targetWorkflowSnapshot,
 		targetWorkflowEventsSeq,
+		gomock.Any(),
 	).Return(nil)
 	targetContext.EXPECT().ReapplyEvents(gomock.Any(), s.mockShard, targetWorkflowEventsSeq).Return(nil)
 
@@ -343,6 +346,7 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_CreateAsZ
 		targetMutableState,
 		targetWorkflowSnapshot,
 		targetWorkflowEventsSeq,
+		gomock.Any(),
 	).Return(nil)
 	targetContext.EXPECT().ReapplyEvents(gomock.Any(), s.mockShard, eventsToApply).Return(nil)
 
@@ -413,6 +417,7 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_CreateAsZ
 		targetMutableState,
 		targetWorkflowSnapshot,
 		targetWorkflowEventsSeq,
+		gomock.Any(),
 	).Return(&persistence.WorkflowConditionFailedError{})
 	targetContext.EXPECT().ReapplyEvents(gomock.Any(), s.mockShard, targetWorkflowEventsSeq).Return(nil)
 

service/history/ndc/workflow_state_replicator_test.go

@@ -189,6 +189,7 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_BrandNew() {
 		gomock.Any(),
 		gomock.Any(),
 		[]*persistence.WorkflowEvents{},
+		gomock.Any(),
 	).Return(nil)
 	s.mockNamespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespace.NewNamespaceForTest(
 		&persistencespb.NamespaceInfo{Name: namespaceName},
@@ -307,6 +308,7 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_Ancestors() {
 		gomock.Any(),
 		gomock.Any(),
 		[]*persistence.WorkflowEvents{},
+		gomock.Any(),
 	).Return(nil)
 	s.mockNamespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespace.NewNamespaceForTest(
 		&persistencespb.NamespaceInfo{Name: namespaceName},

service/history/shard/context_impl.go

@@ -24,6 +24,8 @@ import (
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/archiver"
 	"go.temporal.io/server/common/backoff"
+	"go.temporal.io/server/common/cache"
+	"go.temporal.io/server/common/quotas"
 	cclock "go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/config"
@@ -72,6 +74,9 @@ const (
 	queueMetricUpdateInterval = 5 * time.Minute
 
 	pendingMaxReplicationTaskID = math.MaxInt64
+
+	workflowIDRateLimiterCacheSize = 10000
+	workflowIDRateLimiterCacheTTL  = 60 * time.Second
 )
 
 var (
@@ -149,6 +154,8 @@ type (
 		stateMachineRegistry *hsm.Registry
 
 		chasmRegistry *chasm.Registry
+
+		workflowIDRateLimiters cache.StoppableCache
 	}
 
 	remoteClusterInfo struct {
@@ -2126,6 +2133,7 @@ func newContext(
 		ioSemaphore:             locks.NewPrioritySemaphore(ioConcurrency),
 		stateMachineRegistry:    stateMachineRegistry,
 		chasmRegistry:           chasmRegistry,
+		workflowIDRateLimiters:  cache.New(workflowIDRateLimiterCacheSize, &cache.Options{TTL: workflowIDRateLimiterCacheTTL}),
 	}
 	shardContext.taskKeyManager = newTaskKeyManager(
 		shardContext.taskCategoryRegistry,
@@ -2232,6 +2240,29 @@ func (s *ContextImpl) ChasmRegistry() *chasm.Registry {
 	return s.chasmRegistry
 }
 
+func (s *ContextImpl) WorkflowIDReuseRL(namespaceID namespace.ID, workflowID string) quotas.RateLimiter {
+	rps := s.config.WorkflowIDReuseRate(namespaceID.String())
+	if rps <= 0 {
+		return nil
+	}
+	burst := max(1, int(float64(rps)*s.config.WorkflowIDReuseBurstRatio(namespaceID.String())))
+	key := namespaceID.String() + "/" + workflowID
+	existing := s.workflowIDRateLimiters.Get(key)
+	if existing == nil {
+		rl := quotas.NewRateLimiter(float64(rps), burst)
+		existing, _ = s.workflowIDRateLimiters.PutIfNotExist(key, rl)
+	}
+	rl, ok := existing.(*quotas.RateLimiterImpl)
+	if !ok {
+		// Should never happen; cache only stores *RateLimiterImpl.
+		return nil
+	}
+	if float64(rps) != rl.Rate() || burst != rl.Burst() {
+		rl.SetRateBurst(float64(rps), burst)
+	}
+	return rl
+}
+
 func (s *ContextImpl) GetCachedWorkflowContext(
 	ctx context.Context,
 	namespaceID namespace.ID,

service/history/shard/context_testutil.go

@@ -8,6 +8,7 @@ import (
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/common/cache"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/future"
@@ -150,6 +151,7 @@ func newTestContext(t *resourcetest.Test, eventsCache events.Cache, config Conte
 		namespaceRegistry:       registry,
 		stateMachineRegistry:    hsm.NewRegistry(),
 		chasmRegistry:           chasm.NewRegistry(t.GetLogger()),
+		workflowIDRateLimiters:  cache.New(workflowIDRateLimiterCacheSize, &cache.Options{TTL: workflowIDRateLimiterCacheTTL}),
 		persistenceShardManager: t.GetShardManager(),
 		clientBean:              t.GetClientBean(),
 		saProvider:              t.GetSearchAttributesProvider(),