Skip to content

Commit 8fd6fd3

Browse files
mastermanumastermanu
authored
Sets default WorkflowExecutionTimeout and WorkflowRunTimeout on the request in FrontEnd. (#763)
When starting a Workflow, we auto fill-in "defaults" for Workflow Execution and Run Timeouts in the history service. Unfortunately, some aspects of the code (such as our retry logic) depends on the default value being filled in even when that auto fill-in logic has not yet been executed. This PR ensures that defaults are filled in at Workflow Start Request creation time to avoid this. As an example, this was causing retries to not be executed with MaxAttempts = 0 if the user did not explicitly set the WorkflowExecutionTime. Unit / Integration / Manual testing Minimal risk
1 parent 666dadb commit 8fd6fd3

File tree

6 files changed

+198
-42
lines changed

6 files changed

+198
-42
lines changed

‎common/constants.go‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,22 @@ const (
107107
MaxWorkflowRetentionPeriod = 30 * time.Hour * 24
108108
)
109109

110+
const (
111+
// DefaultWorkflowExecutionTimeout is the Default Workflow Execution timeout applied to a Workflow when
112+
// this value is not explicitly set by the user on a Start Workflow request
113+
// Intention is 10 years
114+
DefaultWorkflowExecutionTimeout = 24 * 365 * 10 * time.Hour
115+
116+
// DefaultWorkflowRunTimeout is the Default Workflow Run timeout applied to a Workflow when
117+
// this value is not explicitly set by the user on a Start Workflow request
118+
// Intention is 10 years
119+
DefaultWorkflowRunTimeout = 24 * 365 * 10 * time.Hour
120+
121+
// DefaultWorkflowTaskTimeout sets the Default Workflow Task timeout for a Workflow
122+
// when the value is not explicitly set by the user. Intention is 10 seconds.
123+
DefaultWorkflowTaskTimeout = 10 * time.Second
124+
)
125+
110126
const (
111127
// DefaultTransactionSizeLimit is the largest allowed transaction size to persistence
112128
DefaultTransactionSizeLimit = 14 * 1024 * 1024

‎common/util.go‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ import (
5050
"go.temporal.io/server/common/metrics"
5151
"go.temporal.io/server/common/payload"
5252
"go.temporal.io/server/common/primitives/timestamp"
53+
"go.temporal.io/server/common/service/dynamicconfig"
5354
serviceerrors "go.temporal.io/server/common/serviceerror"
5455
)
5556

@@ -668,3 +669,54 @@ func GetPayloadsMapSize(data map[string]*commonpb.Payloads) int {
668669

669670
return size
670671
}
672+
673+
// GetWorkflowExecutionTimeout gets the default allowed execution timeout or truncates the requested value to the maximum allowed timeout
674+
func GetWorkflowExecutionTimeout(
675+
namespace string,
676+
requestedTimeout time.Duration,
677+
getDefaultTimeoutFunc dynamicconfig.DurationPropertyFnWithNamespaceFilter,
678+
getMaxAllowedTimeoutFunc dynamicconfig.DurationPropertyFnWithNamespaceFilter) time.Duration {
679+
680+
if requestedTimeout == 0 {
681+
requestedTimeout = getDefaultTimeoutFunc(namespace)
682+
}
683+
684+
return timestamp.MinDuration(
685+
requestedTimeout,
686+
getMaxAllowedTimeoutFunc(namespace),
687+
)
688+
}
689+
690+
// GetWorkflowRunTimeout gets the default allowed run timeout or truncates the requested value to the maximum allowed timeout
691+
func GetWorkflowRunTimeout(
692+
namespace string,
693+
requestedTimeout time.Duration,
694+
executionTimeout time.Duration,
695+
getDefaultTimeoutFunc dynamicconfig.DurationPropertyFnWithNamespaceFilter,
696+
getMaxAllowedTimeoutFunc dynamicconfig.DurationPropertyFnWithNamespaceFilter) time.Duration {
697+
698+
if requestedTimeout == 0 {
699+
requestedTimeout = getDefaultTimeoutFunc(namespace)
700+
}
701+
702+
return timestamp.MinDuration(
703+
timestamp.MinDuration(
704+
requestedTimeout,
705+
executionTimeout,
706+
),
707+
getMaxAllowedTimeoutFunc(namespace),
708+
)
709+
}
710+
711+
// GetWorkflowTaskTimeout gets the default allowed execution timeout or truncates the requested value to the maximum allowed timeout
712+
func GetWorkflowTaskTimeout(
713+
namespace string,
714+
requestedTimeout time.Duration,
715+
getDefaultTimeoutFunc dynamicconfig.DurationPropertyFnWithNamespaceFilter) time.Duration {
716+
717+
if requestedTimeout == 0 {
718+
return getDefaultTimeoutFunc(namespace)
719+
}
720+
721+
return requestedTimeout
722+
}

‎service/frontend/service.go‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,19 @@ type Config struct {
108108

109109
EnableRPCReplication dynamicconfig.BoolPropertyFn
110110
EnableCleanupReplicationTask dynamicconfig.BoolPropertyFn
111+
112+
// The execution timeout a workflow execution defaults to if not specified
113+
DefaultWorkflowExecutionTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter
114+
// The run timeout a workflow run defaults to if not specified
115+
DefaultWorkflowRunTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter
116+
117+
// The execution timeout a workflow execution defaults to if not specified
118+
MaxWorkflowExecutionTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter
119+
// The run timeout a workflow run defaults to if not specified
120+
MaxWorkflowRunTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter
121+
122+
// DefaultWorkflowTaskTimeout the default workflow task timeout
123+
DefaultWorkflowTaskTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter
111124
}
112125

113126
// NewConfig returns new service config with default values
@@ -149,6 +162,11 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int, enableReadFro
149162
EnableRPCReplication: dc.GetBoolProperty(dynamicconfig.FrontendEnableRPCReplication, false),
150163
EnableCleanupReplicationTask: dc.GetBoolProperty(dynamicconfig.FrontendEnableCleanupReplicationTask, true),
151164
DefaultWorkflowRetryPolicy: dc.GetMapPropertyFnWithNamespaceFilter(dynamicconfig.DefaultWorkflowRetryPolicy, common.GetDefaultRetryPolicyConfigOptions()),
165+
DefaultWorkflowExecutionTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowExecutionTimeout, common.DefaultWorkflowExecutionTimeout),
166+
DefaultWorkflowRunTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowRunTimeout, common.DefaultWorkflowRunTimeout),
167+
MaxWorkflowExecutionTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.MaxWorkflowExecutionTimeout, common.DefaultWorkflowExecutionTimeout),
168+
MaxWorkflowRunTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.MaxWorkflowRunTimeout, common.DefaultWorkflowRunTimeout),
169+
DefaultWorkflowTaskTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowTaskTimeout, common.DefaultWorkflowTaskTimeout),
152170
}
153171
}
154172

‎service/frontend/workflowHandler.go‎

Lines changed: 94 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -452,16 +452,8 @@ func (wh *WorkflowHandler) StartWorkflowExecution(ctx context.Context, request *
452452
return nil, err
453453
}
454454

455-
if timestamp.DurationValue(request.GetWorkflowExecutionTimeout()) < 0 {
456-
return nil, wh.error(errInvalidWorkflowExecutionTimeoutSeconds, scope)
457-
}
458-
459-
if timestamp.DurationValue(request.GetWorkflowRunTimeout()) < 0 {
460-
return nil, wh.error(errInvalidWorkflowRunTimeoutSeconds, scope)
461-
}
462-
463-
if timestamp.DurationValue(request.GetWorkflowTaskTimeout()) < 0 {
464-
return nil, wh.error(errInvalidWorkflowTaskTimeoutSeconds, scope)
455+
if err := wh.validateStartWorkflowTimeouts(scope, request); err != nil {
456+
return nil, err
465457
}
466458

467459
if request.GetRequestId() == "" {
@@ -2144,16 +2136,8 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(ctx context.Context,
21442136
return nil, wh.error(errRequestIDTooLong, scope)
21452137
}
21462138

2147-
if timestamp.DurationValue(request.GetWorkflowExecutionTimeout()) < 0 {
2148-
return nil, wh.error(errInvalidWorkflowExecutionTimeoutSeconds, scope)
2149-
}
2150-
2151-
if timestamp.DurationValue(request.GetWorkflowRunTimeout()) < 0 {
2152-
return nil, wh.error(errInvalidWorkflowRunTimeoutSeconds, scope)
2153-
}
2154-
2155-
if timestamp.DurationValue(request.GetWorkflowTaskTimeout()) < 0 {
2156-
return nil, wh.error(errInvalidWorkflowTaskTimeoutSeconds, scope)
2139+
if err := wh.validateSignalWithStartWorkflowTimeouts(scope, request); err != nil {
2140+
return nil, err
21572141
}
21582142

21592143
if err := wh.validateRetryPolicy(request.GetNamespace(), request.RetryPolicy); err != nil {
@@ -3755,3 +3739,93 @@ func (wh *WorkflowHandler) validateRetryPolicy(namespace string, retryPolicy *co
37553739
common.EnsureRetryPolicyDefaults(retryPolicy, defaultWorkflowRetrySettings)
37563740
return common.ValidateRetryPolicy(retryPolicy)
37573741
}
3742+
3743+
func (wh *WorkflowHandler) validateStartWorkflowTimeouts(
3744+
scope metrics.Scope,
3745+
request *workflowservice.StartWorkflowExecutionRequest) error {
3746+
if timestamp.DurationValue(request.GetWorkflowExecutionTimeout()) < 0 {
3747+
return wh.error(errInvalidWorkflowExecutionTimeoutSeconds, scope)
3748+
}
3749+
3750+
if timestamp.DurationValue(request.GetWorkflowRunTimeout()) < 0 {
3751+
return wh.error(errInvalidWorkflowRunTimeoutSeconds, scope)
3752+
}
3753+
3754+
if timestamp.DurationValue(request.GetWorkflowTaskTimeout()) < 0 {
3755+
return wh.error(errInvalidWorkflowTaskTimeoutSeconds, scope)
3756+
}
3757+
3758+
request.WorkflowExecutionTimeout = timestamp.DurationPtr(
3759+
common.GetWorkflowExecutionTimeout(
3760+
request.GetNamespace(),
3761+
timestamp.DurationValue(request.GetWorkflowExecutionTimeout()),
3762+
wh.config.DefaultWorkflowExecutionTimeout,
3763+
wh.config.MaxWorkflowExecutionTimeout,
3764+
),
3765+
)
3766+
3767+
request.WorkflowRunTimeout = timestamp.DurationPtr(
3768+
common.GetWorkflowRunTimeout(
3769+
request.GetNamespace(),
3770+
timestamp.DurationValue(request.GetWorkflowRunTimeout()),
3771+
timestamp.DurationValue(request.GetWorkflowExecutionTimeout()),
3772+
wh.config.DefaultWorkflowRunTimeout,
3773+
wh.config.MaxWorkflowRunTimeout,
3774+
),
3775+
)
3776+
3777+
request.WorkflowTaskTimeout = timestamp.DurationPtr(
3778+
common.GetWorkflowTaskTimeout(
3779+
request.GetNamespace(),
3780+
timestamp.DurationValue(request.GetWorkflowTaskTimeout()),
3781+
wh.config.DefaultWorkflowTaskTimeout,
3782+
),
3783+
)
3784+
3785+
return nil
3786+
}
3787+
3788+
func (wh *WorkflowHandler) validateSignalWithStartWorkflowTimeouts(
3789+
scope metrics.Scope,
3790+
request *workflowservice.SignalWithStartWorkflowExecutionRequest) error {
3791+
if timestamp.DurationValue(request.GetWorkflowExecutionTimeout()) < 0 {
3792+
return wh.error(errInvalidWorkflowExecutionTimeoutSeconds, scope)
3793+
}
3794+
3795+
if timestamp.DurationValue(request.GetWorkflowRunTimeout()) < 0 {
3796+
return wh.error(errInvalidWorkflowRunTimeoutSeconds, scope)
3797+
}
3798+
3799+
if timestamp.DurationValue(request.GetWorkflowTaskTimeout()) < 0 {
3800+
return wh.error(errInvalidWorkflowTaskTimeoutSeconds, scope)
3801+
}
3802+
3803+
request.WorkflowExecutionTimeout = timestamp.DurationPtr(
3804+
common.GetWorkflowExecutionTimeout(
3805+
request.GetNamespace(),
3806+
timestamp.DurationValue(request.GetWorkflowExecutionTimeout()),
3807+
wh.config.DefaultWorkflowExecutionTimeout,
3808+
wh.config.MaxWorkflowExecutionTimeout,
3809+
),
3810+
)
3811+
3812+
request.WorkflowRunTimeout = timestamp.DurationPtr(
3813+
common.GetWorkflowRunTimeout(
3814+
request.GetNamespace(),
3815+
timestamp.DurationValue(request.GetWorkflowRunTimeout()),
3816+
timestamp.DurationValue(request.GetWorkflowExecutionTimeout()),
3817+
wh.config.DefaultWorkflowRunTimeout,
3818+
wh.config.MaxWorkflowRunTimeout,
3819+
),
3820+
)
3821+
3822+
request.WorkflowTaskTimeout = timestamp.DurationPtr(
3823+
common.GetWorkflowTaskTimeout(
3824+
request.GetNamespace(),
3825+
timestamp.DurationValue(request.GetWorkflowTaskTimeout()),
3826+
wh.config.DefaultWorkflowTaskTimeout,
3827+
),
3828+
)
3829+
3830+
return nil
3831+
}

‎service/history/service.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -273,7 +273,7 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int, storeType strin
273273
VisibilityOpenMaxQPS: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryVisibilityOpenMaxQPS, 300),
274274
VisibilityClosedMaxQPS: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryVisibilityClosedMaxQPS, 300),
275275
MaxAutoResetPoints: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryMaxAutoResetPoints, defaultHistoryMaxAutoResetPoints),
276-
DefaultWorkflowTaskTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowTaskTimeout, time.Second*10),
276+
DefaultWorkflowTaskTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowTaskTimeout, common.DefaultWorkflowTaskTimeout),
277277
MaxWorkflowTaskTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.MaxWorkflowTaskTimeout, time.Second*60),
278278
AdvancedVisibilityWritingMode: dc.GetStringProperty(dynamicconfig.AdvancedVisibilityWritingMode, common.GetDefaultAdvancedVisibilityWritingMode(isAdvancedVisConfigExist)),
279279
EmitShardDiffLog: dc.GetBoolProperty(dynamicconfig.EmitShardDiffLog, false),
@@ -387,10 +387,10 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int, storeType strin
387387
SearchAttributesTotalSizeLimit: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.SearchAttributesTotalSizeLimit, 40*1024),
388388
StickyTTL: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.StickyTTL, time.Hour*24*365),
389389
WorkflowTaskHeartbeatTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.WorkflowTaskHeartbeatTimeout, time.Minute*30),
390-
DefaultWorkflowExecutionTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowExecutionTimeout, time.Hour*24*365*10),
391-
DefaultWorkflowRunTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowRunTimeout, time.Hour*24*365*10),
392-
MaxWorkflowExecutionTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.MaxWorkflowExecutionTimeout, time.Hour*24*365*10),
393-
MaxWorkflowRunTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.MaxWorkflowRunTimeout, time.Hour*24*365*10),
390+
DefaultWorkflowExecutionTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowExecutionTimeout, common.DefaultWorkflowExecutionTimeout),
391+
DefaultWorkflowRunTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowRunTimeout, common.DefaultWorkflowRunTimeout),
392+
MaxWorkflowExecutionTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.MaxWorkflowExecutionTimeout, common.DefaultWorkflowExecutionTimeout),
393+
MaxWorkflowRunTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.MaxWorkflowRunTimeout, common.DefaultWorkflowRunTimeout),
394394
ReplicationTaskFetcherParallelism: dc.GetIntProperty(dynamicconfig.ReplicationTaskFetcherParallelism, 1),
395395
ReplicationTaskFetcherAggregationInterval: dc.GetDurationProperty(dynamicconfig.ReplicationTaskFetcherAggregationInterval, 2*time.Second),
396396
ReplicationTaskFetcherTimerJitterCoefficient: dc.GetFloat64Property(dynamicconfig.ReplicationTaskFetcherTimerJitterCoefficient, 0.15),

‎service/history/workflowExecutionUtil.go‎

Lines changed: 13 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -230,26 +230,22 @@ func terminateWorkflow(
230230
}
231231

232232
func getWorkflowExecutionTimeout(namespace string, requestedTimeout time.Duration, serviceConfig *Config) time.Duration {
233-
executionTimeoutSeconds := requestedTimeout
234-
if executionTimeoutSeconds == 0 {
235-
executionTimeoutSeconds = timestamp.RoundUp(serviceConfig.DefaultWorkflowExecutionTimeout(namespace))
236-
}
237-
maxWorkflowExecutionTimeout := timestamp.RoundUp(serviceConfig.MaxWorkflowExecutionTimeout(namespace))
238-
executionTimeoutSeconds = timestamp.MinDuration(executionTimeoutSeconds, maxWorkflowExecutionTimeout)
239-
240-
return executionTimeoutSeconds
233+
return common.GetWorkflowExecutionTimeout(
234+
namespace,
235+
requestedTimeout,
236+
serviceConfig.DefaultWorkflowExecutionTimeout,
237+
serviceConfig.MaxWorkflowExecutionTimeout,
238+
)
241239
}
242240

243241
func getWorkflowRunTimeout(namespace string, requestedTimeout, executionTimeout time.Duration, serviceConfig *Config) time.Duration {
244-
runTimeoutSeconds := requestedTimeout
245-
if runTimeoutSeconds == 0 {
246-
runTimeoutSeconds = timestamp.RoundUp(serviceConfig.DefaultWorkflowRunTimeout(namespace))
247-
}
248-
maxWorkflowRunTimeout := timestamp.RoundUp(serviceConfig.MaxWorkflowRunTimeout(namespace))
249-
runTimeoutSeconds = timestamp.MinDuration(runTimeoutSeconds, maxWorkflowRunTimeout)
250-
runTimeoutSeconds = timestamp.MinDuration(runTimeoutSeconds, executionTimeout)
251-
252-
return runTimeoutSeconds
242+
return common.GetWorkflowRunTimeout(
243+
namespace,
244+
requestedTimeout,
245+
executionTimeout,
246+
serviceConfig.DefaultWorkflowRunTimeout,
247+
serviceConfig.MaxWorkflowRunTimeout,
248+
)
253249
}
254250

255251
// FindAutoResetPoint returns the auto reset point

0 commit comments

Comments
 (0)