Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Remove FollowReusePolicyAfterConflictPolicyTerminate (#7810)

Browse Source 
## What changed?

Removed config option `FollowReusePolicyAfterConflictPolicyTerminate`.

## Why?

It was [put in place](https://github.com/temporalio/temporal/pull/7099)
to have the ability to roll back a behavior change. It's been 4+ months
and it's okay to remove it now.

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)
This commit is contained in:
Stephan Behnke
2025-05-28 13:17:00 -07:00
committed by GitHub
parent 4d4f0693a8
commit a6cf57cb81
6 changed files with 92 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
common/dynamicconfig/constants.go
View File

@@ -2121,13 +2121,6 @@ the user has not specified an explicit RetryPolicy`,
retrypolicy.DefaultDefaultRetrySettings,
`DefaultWorkflowRetryPolicy represents the out-of-box retry policy for unset fields
where the user has set an explicit RetryPolicy, but not specified all the fields`,
)
FollowReusePolicyAfterConflictPolicyTerminate = NewNamespaceBoolSetting(
"history.followReusePolicyAfterConflictPolicyTerminate",
true,
`Follows WorkflowIdReusePolicy RejectDuplicate and AllowDuplicateFailedOnly after WorkflowIdReusePolicy TerminateExisting was applied.
If true (the default), RejectDuplicate is disallowed and AllowDuplicateFailedOnly will be honored after TerminateExisting is applied.
This configuration will be become the default behavior in the next release and removed subsequently.`,
)
AllowResetWithPendingChildren = NewNamespaceBoolSetting(
"history.allowResetWithPendingChildren",

67
service/frontend/service.go
View File

@@ -107,8 +107,6 @@ type Config struct {
// specified RetryPolicy
DefaultWorkflowRetryPolicy dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
FollowReusePolicyAfterConflictPolicyTerminate dynamicconfig.TypedPropertyFnWithNamespaceFilter[bool]
// VisibilityArchival system protection
VisibilityArchivalQueryMaxPageSize dynamicconfig.IntPropertyFn
@@ -266,39 +264,38 @@ func NewConfig(
// Overshoot since these low rate limits don't work well in an uncoordinated global limiter.
GlobalNamespaceNamespaceReplicationInducingAPIsRPS: dynamicconfig.FrontendGlobalNamespaceNamespaceReplicationInducingAPIsRPS.Get(dc),
MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc),
WorkerBuildIdSizeLimit: dynamicconfig.WorkerBuildIdSizeLimit.Get(dc),
ReachabilityTaskQueueScanLimit: dynamicconfig.ReachabilityTaskQueueScanLimit.Get(dc),
ReachabilityQueryBuildIdLimit: dynamicconfig.ReachabilityQueryBuildIdLimit.Get(dc),
ReachabilityCacheOpenWFsTTL: dynamicconfig.ReachabilityCacheOpenWFsTTL.Get(dc),
ReachabilityCacheClosedWFsTTL: dynamicconfig.ReachabilityCacheClosedWFsTTL.Get(dc),
ReachabilityQuerySetDurationSinceDefault: dynamicconfig.ReachabilityQuerySetDurationSinceDefault.Get(dc),
MaxBadBinaries: dynamicconfig.FrontendMaxBadBinaries.Get(dc),
DisableListVisibilityByFilter: dynamicconfig.DisableListVisibilityByFilter.Get(dc),
BlobSizeLimitError: dynamicconfig.BlobSizeLimitError.Get(dc),
BlobSizeLimitWarn: dynamicconfig.BlobSizeLimitWarn.Get(dc),
ThrottledLogRPS: dynamicconfig.FrontendThrottledLogRPS.Get(dc),
ShutdownDrainDuration: dynamicconfig.FrontendShutdownDrainDuration.Get(dc),
ShutdownFailHealthCheckDuration: dynamicconfig.FrontendShutdownFailHealthCheckDuration.Get(dc),
EnableNamespaceNotActiveAutoForwarding: dynamicconfig.EnableNamespaceNotActiveAutoForwarding.Get(dc),
SearchAttributesNumberOfKeysLimit: dynamicconfig.SearchAttributesNumberOfKeysLimit.Get(dc),
SearchAttributesSizeOfValueLimit: dynamicconfig.SearchAttributesSizeOfValueLimit.Get(dc),
SearchAttributesTotalSizeLimit: dynamicconfig.SearchAttributesTotalSizeLimit.Get(dc),
VisibilityArchivalQueryMaxPageSize: dynamicconfig.VisibilityArchivalQueryMaxPageSize.Get(dc),
DisallowQuery: dynamicconfig.DisallowQuery.Get(dc),
SendRawWorkflowHistory: dynamicconfig.SendRawWorkflowHistory.Get(dc),
DefaultWorkflowRetryPolicy: dynamicconfig.DefaultWorkflowRetryPolicy.Get(dc),
FollowReusePolicyAfterConflictPolicyTerminate: dynamicconfig.FollowReusePolicyAfterConflictPolicyTerminate.Get(dc),
DefaultWorkflowTaskTimeout: dynamicconfig.DefaultWorkflowTaskTimeout.Get(dc),
EnableServerVersionCheck: dynamicconfig.EnableServerVersionCheck.Get(dc),
EnableTokenNamespaceEnforcement: dynamicconfig.EnableTokenNamespaceEnforcement.Get(dc),
KeepAliveMinTime: dynamicconfig.KeepAliveMinTime.Get(dc),
KeepAlivePermitWithoutStream: dynamicconfig.KeepAlivePermitWithoutStream.Get(dc),
KeepAliveMaxConnectionIdle: dynamicconfig.KeepAliveMaxConnectionIdle.Get(dc),
KeepAliveMaxConnectionAge: dynamicconfig.KeepAliveMaxConnectionAge.Get(dc),
KeepAliveMaxConnectionAgeGrace: dynamicconfig.KeepAliveMaxConnectionAgeGrace.Get(dc),
KeepAliveTime: dynamicconfig.KeepAliveTime.Get(dc),
KeepAliveTimeout: dynamicconfig.KeepAliveTimeout.Get(dc),
MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc),
WorkerBuildIdSizeLimit: dynamicconfig.WorkerBuildIdSizeLimit.Get(dc),
ReachabilityTaskQueueScanLimit: dynamicconfig.ReachabilityTaskQueueScanLimit.Get(dc),
ReachabilityQueryBuildIdLimit: dynamicconfig.ReachabilityQueryBuildIdLimit.Get(dc),
ReachabilityCacheOpenWFsTTL: dynamicconfig.ReachabilityCacheOpenWFsTTL.Get(dc),
ReachabilityCacheClosedWFsTTL: dynamicconfig.ReachabilityCacheClosedWFsTTL.Get(dc),
ReachabilityQuerySetDurationSinceDefault: dynamicconfig.ReachabilityQuerySetDurationSinceDefault.Get(dc),
MaxBadBinaries: dynamicconfig.FrontendMaxBadBinaries.Get(dc),
DisableListVisibilityByFilter: dynamicconfig.DisableListVisibilityByFilter.Get(dc),
BlobSizeLimitError: dynamicconfig.BlobSizeLimitError.Get(dc),
BlobSizeLimitWarn: dynamicconfig.BlobSizeLimitWarn.Get(dc),
ThrottledLogRPS: dynamicconfig.FrontendThrottledLogRPS.Get(dc),
ShutdownDrainDuration: dynamicconfig.FrontendShutdownDrainDuration.Get(dc),
ShutdownFailHealthCheckDuration: dynamicconfig.FrontendShutdownFailHealthCheckDuration.Get(dc),
EnableNamespaceNotActiveAutoForwarding: dynamicconfig.EnableNamespaceNotActiveAutoForwarding.Get(dc),
SearchAttributesNumberOfKeysLimit: dynamicconfig.SearchAttributesNumberOfKeysLimit.Get(dc),
SearchAttributesSizeOfValueLimit: dynamicconfig.SearchAttributesSizeOfValueLimit.Get(dc),
SearchAttributesTotalSizeLimit: dynamicconfig.SearchAttributesTotalSizeLimit.Get(dc),
VisibilityArchivalQueryMaxPageSize: dynamicconfig.VisibilityArchivalQueryMaxPageSize.Get(dc),
DisallowQuery: dynamicconfig.DisallowQuery.Get(dc),
SendRawWorkflowHistory: dynamicconfig.SendRawWorkflowHistory.Get(dc),
DefaultWorkflowRetryPolicy: dynamicconfig.DefaultWorkflowRetryPolicy.Get(dc),
DefaultWorkflowTaskTimeout: dynamicconfig.DefaultWorkflowTaskTimeout.Get(dc),
EnableServerVersionCheck: dynamicconfig.EnableServerVersionCheck.Get(dc),
EnableTokenNamespaceEnforcement: dynamicconfig.EnableTokenNamespaceEnforcement.Get(dc),
KeepAliveMinTime: dynamicconfig.KeepAliveMinTime.Get(dc),
KeepAlivePermitWithoutStream: dynamicconfig.KeepAlivePermitWithoutStream.Get(dc),
KeepAliveMaxConnectionIdle: dynamicconfig.KeepAliveMaxConnectionIdle.Get(dc),
KeepAliveMaxConnectionAge: dynamicconfig.KeepAliveMaxConnectionAge.Get(dc),
KeepAliveMaxConnectionAgeGrace: dynamicconfig.KeepAliveMaxConnectionAgeGrace.Get(dc),
KeepAliveTime: dynamicconfig.KeepAliveTime.Get(dc),
KeepAliveTimeout: dynamicconfig.KeepAliveTimeout.Get(dc),
DeleteNamespaceDeleteActivityRPS: dynamicconfig.DeleteNamespaceDeleteActivityRPS.Get(dc),
DeleteNamespacePageSize: dynamicconfig.DeleteNamespacePageSize.Get(dc),

95
service/frontend/workflow_handler.go
View File

@@ -98,36 +98,35 @@ type (
workflowservice.UnimplementedWorkflowServiceServer
status int32
tokenSerializer *tasktoken.Serializer
config *Config
versionChecker headers.VersionChecker
namespaceHandler *namespaceHandler
getDefaultWorkflowRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
followReusePolicyAfterConflictPolicyTerminate dynamicconfig.TypedPropertyFnWithNamespaceFilter[bool]
visibilityMgr manager.VisibilityManager
logger log.Logger
throttledLogger log.Logger
persistenceExecutionName string
clusterMetadataManager persistence.ClusterMetadataManager
clusterMetadata cluster.Metadata
historyClient historyservice.HistoryServiceClient
matchingClient matchingservice.MatchingServiceClient
deploymentStoreClient deployment.DeploymentStoreClient
workerDeploymentClient workerdeployment.Client
archiverProvider provider.ArchiverProvider
payloadSerializer serialization.Serializer
namespaceRegistry namespace.Registry
saMapperProvider searchattribute.MapperProvider
saProvider searchattribute.Provider
saValidator *searchattribute.Validator
archivalMetadata archiver.ArchivalMetadata
healthServer *health.Server
overrides *Overrides
membershipMonitor membership.Monitor
healthInterceptor *interceptor.HealthInterceptor
scheduleSpecBuilder *scheduler.SpecBuilder
outstandingPollers collection.SyncMap[string, collection.SyncMap[string, context.CancelFunc]]
httpEnabled bool
tokenSerializer *tasktoken.Serializer
config *Config
versionChecker headers.VersionChecker
namespaceHandler *namespaceHandler
getDefaultWorkflowRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
visibilityMgr manager.VisibilityManager
logger log.Logger
throttledLogger log.Logger
persistenceExecutionName string
clusterMetadataManager persistence.ClusterMetadataManager
clusterMetadata cluster.Metadata
historyClient historyservice.HistoryServiceClient
matchingClient matchingservice.MatchingServiceClient
deploymentStoreClient deployment.DeploymentStoreClient
workerDeploymentClient workerdeployment.Client
archiverProvider provider.ArchiverProvider
payloadSerializer serialization.Serializer
namespaceRegistry namespace.Registry
saMapperProvider searchattribute.MapperProvider
saProvider searchattribute.Provider
saValidator *searchattribute.Validator
archivalMetadata archiver.ArchivalMetadata
healthServer *health.Server
overrides *Overrides
membershipMonitor membership.Monitor
healthInterceptor *interceptor.HealthInterceptor
scheduleSpecBuilder *scheduler.SpecBuilder
outstandingPollers collection.SyncMap[string, collection.SyncMap[string, context.CancelFunc]]
httpEnabled bool
}
)
@@ -174,23 +173,22 @@ func NewWorkflowHandler(
timeSource,
config,
),
getDefaultWorkflowRetrySettings: config.DefaultWorkflowRetryPolicy,
followReusePolicyAfterConflictPolicyTerminate: config.FollowReusePolicyAfterConflictPolicyTerminate,
visibilityMgr: visibilityMgr,
logger: logger,
throttledLogger: throttledLogger,
persistenceExecutionName: persistenceExecutionName,
clusterMetadataManager: clusterMetadataManager,
clusterMetadata: clusterMetadata,
historyClient: historyClient,
matchingClient: matchingClient,
deploymentStoreClient: deploymentStoreClient,
workerDeploymentClient: workerDeploymentClient,
archiverProvider: archiverProvider,
payloadSerializer: payloadSerializer,
namespaceRegistry: namespaceRegistry,
saProvider: saProvider,
saMapperProvider: saMapperProvider,
getDefaultWorkflowRetrySettings: config.DefaultWorkflowRetryPolicy,
visibilityMgr: visibilityMgr,
logger: logger,
throttledLogger: throttledLogger,
persistenceExecutionName: persistenceExecutionName,
clusterMetadataManager: clusterMetadataManager,
clusterMetadata: clusterMetadata,
historyClient: historyClient,
matchingClient: matchingClient,
deploymentStoreClient: deploymentStoreClient,
workerDeploymentClient: workerDeploymentClient,
archiverProvider: archiverProvider,
payloadSerializer: payloadSerializer,
namespaceRegistry: namespaceRegistry,
saProvider: saProvider,
saMapperProvider: saMapperProvider,
saValidator: searchattribute.NewValidator(
saProvider,
saMapperProvider,
@@ -5169,8 +5167,7 @@ func (wh *WorkflowHandler) validateWorkflowIdReusePolicy(
return errIncompatibleIDReusePolicyTerminateIfRunning
}
if conflictPolicy == enumspb.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING &&
reusePolicy == enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE &&
wh.followReusePolicyAfterConflictPolicyTerminate(namespaceName.String()) {
reusePolicy == enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE {
return errIncompatibleIDReusePolicyRejectDuplicate
}
return nil

11
service/frontend/workflow_handler_test.go
View File

@@ -639,23 +639,12 @@ func (s *WorkflowHandlerSuite) TestStartWorkflowExecution_InvalidWorkflowIdReuse
WorkflowIdConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING,
}
// by default, disallow
config := s.newConfig()
wh := s.getWorkflowHandler(config)
resp, err := wh.StartWorkflowExecution(context.Background(), req)
s.Nil(resp)
s.Equal(err, serviceerror.NewInvalidArgument(
"Invalid WorkflowIDReusePolicy: WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE cannot be used together with WorkflowIdConflictPolicy WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING"))
// allow if explicitly allowed
s.mockSearchAttributesMapperProvider.EXPECT().GetMapper(gomock.Any()).Return(nil, nil)
s.mockNamespaceCache.EXPECT().GetNamespaceID(gomock.Any()).Return(namespace.NewID(), nil)
s.mockHistoryClient.EXPECT().StartWorkflowExecution(gomock.Any(), gomock.Any()).Return(&historyservice.StartWorkflowExecutionResponse{Started: true}, nil)
config.FollowReusePolicyAfterConflictPolicyTerminate = dc.GetBoolPropertyFnFilteredByNamespace(false)
wh = s.getWorkflowHandler(config)
_, err = wh.StartWorkflowExecution(context.Background(), req)
s.NoError(err)
}
func (s *WorkflowHandlerSuite) TestStartWorkflowExecution_DefaultWorkflowIdDuplicationPolicies() {

34
service/history/api/startworkflow/api.go
View File

@@ -49,16 +49,15 @@ const (
// Starter starts a new workflow execution.
type Starter struct {
metricsHandler metrics.Handler
shardContext historyi.ShardContext
workflowConsistencyChecker api.WorkflowConsistencyChecker
tokenSerializer *tasktoken.Serializer
visibilityManager manager.VisibilityManager
request *historyservice.StartWorkflowExecutionRequest
namespace *namespace.Namespace
createOrUpdateLeaseFn api.CreateOrUpdateLeaseFunc
followReusePolicyAfterConflictPolicyTerminate dynamicconfig.BoolPropertyFnWithNamespaceFilter
enableRequestIdRefLinks dynamicconfig.BoolPropertyFn
metricsHandler metrics.Handler
shardContext historyi.ShardContext
workflowConsistencyChecker api.WorkflowConsistencyChecker
tokenSerializer *tasktoken.Serializer
visibilityManager manager.VisibilityManager
request *historyservice.StartWorkflowExecutionRequest
namespace *namespace.Namespace
createOrUpdateLeaseFn api.CreateOrUpdateLeaseFunc
enableRequestIdRefLinks dynamicconfig.BoolPropertyFn
}
// creationParams is a container for all information obtained from creating the uncommitted execution.
@@ -104,8 +103,7 @@ func NewStarter(
request: request,
namespace: namespaceEntry,
createOrUpdateLeaseFn: createLeaseFn,
followReusePolicyAfterConflictPolicyTerminate: shardContext.GetConfig().FollowReusePolicyAfterConflictPolicyTerminate,
enableRequestIdRefLinks: shardContext.GetConfig().EnableRequestIdRefLinks,
enableRequestIdRefLinks: shardContext.GetConfig().EnableRequestIdRefLinks,
}, nil
}
@@ -491,14 +489,10 @@ func (s *Starter) resolveDuplicateWorkflowID(
resp, err := s.generateResponse(newRunID, mutableStateInfo.workflowTask, events)
return resp, StartNew, err
case consts.ErrWorkflowCompleted:
if s.followReusePolicyAfterConflictPolicyTerminate(s.namespace.Name().String()) {
// Exit and retry again from the top.
// By returning an Unavailable service error, the entire Start request will be retried.
// NOTE: This WorkflowIDReusePolicy cannot be RejectDuplicate as the frontend will reject that.
return nil, StartErr, serviceerror.NewUnavailablef("Termination failed: %v", err)
}
// Fallthough to the logic for only creating the new workflow below.
return nil, StartNew, nil
// Exit and retry again from the top.
// By returning an Unavailable service error, the entire Start request will be retried.
// NOTE: This WorkflowIDReusePolicy cannot be RejectDuplicate as the frontend will reject that.
return nil, StartErr, serviceerror.NewUnavailablef("Termination failed: %v", err)
default:
return nil, StartErr, err
}

4
service/history/configs/config.go
View File

@@ -330,8 +330,6 @@ type Config struct {
EnableEagerWorkflowStart dynamicconfig.BoolPropertyFnWithNamespaceFilter
NamespaceCacheRefreshInterval dynamicconfig.DurationPropertyFn
FollowReusePolicyAfterConflictPolicyTerminate dynamicconfig.BoolPropertyFnWithNamespaceFilter
// ArchivalQueueProcessor settings
ArchivalProcessorSchedulerWorkerCount dynamicconfig.TypedSubscribable[int]
ArchivalProcessorMaxPollHostRPS dynamicconfig.IntPropertyFn
@@ -542,8 +540,6 @@ func NewConfig(
ReplicationProgressCacheMaxSize: dynamicconfig.ReplicationProgressCacheMaxSize.Get(dc),
ReplicationProgressCacheTTL: dynamicconfig.ReplicationProgressCacheTTL.Get(dc),
FollowReusePolicyAfterConflictPolicyTerminate: dynamicconfig.FollowReusePolicyAfterConflictPolicyTerminate.Get(dc),
MaximumBufferedEventsBatch: dynamicconfig.MaximumBufferedEventsBatch.Get(dc),
MaximumBufferedEventsSizeInBytes: dynamicconfig.MaximumBufferedEventsSizeInBytes.Get(dc),
MaximumSignalsPerExecution: dynamicconfig.MaximumSignalsPerExecution.Get(dc),