Adds automatic population of unset retrypolicy field when starting Workflows (#654)

Change summary:

Adds a separate dynamic configuration to dictate "default" retry policy field values for Workflows vs. Activities
Adds logic to auto-populate unset retry policy fields for starting a workflow, signaling-to-starting a workflow, and starting a child workflow. If the specified retry policy is nil, no auto-population happens. If it is non-nil, unset fields are assigned defaults
Fixes bug where WorkflowExpirationTime was always set to 0 for Child Workflows. This meant that Child Workflows would assume that max attempts = 0 implies no retry policy is set.
Addresses flakiness in matching test (#243)

Author: mastermanu

common/defaultActivityRetrySettings.go

@@ -0,0 +1,13 @@
+package common
+
+import "time"
+
+// DefaultRetrySettings indicates what the "default" retry settings
+// are if it is not specified on an Activity or for any unset fields
+// if a policy is explicitly set on a workflow
+type DefaultRetrySettings struct {
+	InitialInterval            time.Duration
+	MaximumIntervalCoefficient float64
+	BackoffCoefficient         float64
+	MaximumAttempts            int32
+}

common/defaultRetrySettings.go

@@ -128,6 +128,9 @@ type MapPropertyFn func(opts ...FilterOption) map[string]interface{}
 // StringPropertyFnWithNamespaceFilter is a wrapper to get string property from dynamic config
 type StringPropertyFnWithNamespaceFilter func(namespace string) string
 
+// MapPropertyFnWithNamespaceFilter is a wrapper to get map property from dynamic config
+type MapPropertyFnWithNamespaceFilter func(namespace string) map[string]interface{}
+
 // BoolPropertyFnWithNamespaceFilter is a wrapper to get bool property from dynamic config
 type BoolPropertyFnWithNamespaceFilter func(namespace string) bool
 
@@ -358,6 +361,18 @@ func (c *Collection) GetStringPropertyFnWithNamespaceFilter(key Key, defaultValu
 	}
 }
 
+// GetMapPropertyFnWithNamespaceFilter gets property and asserts that it's a map
+func (c *Collection) GetMapPropertyFnWithNamespaceFilter(key Key, defaultValue map[string]interface{}) MapPropertyFnWithNamespaceFilter {
+	return func(namespace string) map[string]interface{} {
+		val, err := c.client.GetMapValue(key, getFilterMap(NamespaceFilter(namespace)), defaultValue)
+		if err != nil {
+			c.logError(key, err)
+		}
+		c.logValue(key, val, defaultValue, reflect.DeepEqual)
+		return val
+	}
+}
+
 // GetBoolPropertyFnWithNamespaceFilter gets property with namespace filter and asserts that its namespace
 func (c *Collection) GetBoolPropertyFnWithNamespaceFilter(key Key, defaultValue bool) BoolPropertyFnWithNamespaceFilter {
 	return func(namespace string) bool {

common/service/dynamicconfig/config.go

@@ -86,3 +86,8 @@ func GetStringPropertyFn(value string) func(opts ...FilterOption) string {
 func GetMapPropertyFn(value map[string]interface{}) func(opts ...FilterOption) map[string]interface{} {
 	return func(...FilterOption) map[string]interface{} { return value }
 }
+
+// GetMapPropertyFnWithNamespaceFilter returns value as MapPropertyFn
+func GetMapPropertyFnWithNamespaceFilter(value map[string]interface{}) func(namespace string) map[string]interface{} {
+	return func(namespace string) map[string]interface{} { return value }
+}

common/service/dynamicconfig/config_mock.go

@@ -258,6 +258,7 @@ var keys = map[Key]string{
 	EnableDropStuckTaskByNamespaceID:                       "history.DropStuckTaskByNamespace",
 	SkipReapplicationByNamespaceId:                         "history.SkipReapplicationByNamespaceId",
 	DefaultActivityRetryPolicy:                             "history.defaultActivityRetryPolicy",
+	DefaultWorkflowRetryPolicy:                             "history.defaultWorkflowRetryPolicy",
 
 	WorkerPersistenceMaxQPS:                         "worker.persistenceMaxQPS",
 	WorkerPersistenceGlobalMaxQPS:                   "worker.persistenceGlobalMaxQPS",
@@ -635,6 +636,9 @@ const (
 	// DefaultActivityRetryPolicy represents the out-of-box retry policy for activities where
 	// the user has not specified an explicit RetryPolicy
 	DefaultActivityRetryPolicy
+	// 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
+	DefaultWorkflowRetryPolicy
 
 	// EnableAdminProtection is whether to enable admin checking
 	EnableAdminProtection

common/service/dynamicconfig/constants.go

@@ -40,6 +40,7 @@ import (
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
+	workflowspb "go.temporal.io/server/api/workflow/v1"
 
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
@@ -79,6 +80,16 @@ const (
 	retryKafkaOperationMaxInterval        = 10 * time.Second
 	retryKafkaOperationExpirationInterval = 30 * time.Second
 
+	defaultInitialInterval            = time.Second
+	defaultMaximumIntervalCoefficient = 100.0
+	defaultBackoffCoefficient         = 2.0
+	defaultMaximumAttempts            = 0
+
+	initialIntervalInSecondsConfigKey   = "InitialIntervalInSeconds"
+	maximumIntervalCoefficientConfigKey = "MaximumIntervalCoefficient"
+	backoffCoefficientConfigKey         = "BackoffCoefficient"
+	maximumAttemptsConfigKey            = "MaximumAttempts"
+
 	contextExpireThreshold = 10 * time.Millisecond
 
 	// FailureReasonCompleteResultExceedsLimit is failureReason for complete result exceeds limit
@@ -370,13 +381,13 @@ func SortInt64Slice(slice []int64) {
 }
 
 // EnsureRetryPolicyDefaults ensures the policy subfields, if not explicitly set, are set to the specified defaults
-func EnsureRetryPolicyDefaults(originalPolicy *commonpb.RetryPolicy, defaultSettings DefaultActivityRetrySettings) {
+func EnsureRetryPolicyDefaults(originalPolicy *commonpb.RetryPolicy, defaultSettings DefaultRetrySettings) {
 	if originalPolicy.GetMaximumAttempts() == 0 {
 		originalPolicy.MaximumAttempts = defaultSettings.MaximumAttempts
 	}
 
 	if timestamp.DurationValue(originalPolicy.GetInitialInterval()) == 0 {
-		originalPolicy.InitialInterval = timestamp.DurationPtr(time.Duration(defaultSettings.InitialIntervalInSeconds) * time.Second)
+		originalPolicy.InitialInterval = timestamp.DurationPtr(defaultSettings.InitialInterval)
 	}
 
 	if timestamp.DurationValue(originalPolicy.GetMaximumInterval()) == 0 {
@@ -394,6 +405,7 @@ func ValidateRetryPolicy(policy *commonpb.RetryPolicy) error {
 		// nil policy is valid which means no retry
 		return nil
 	}
+
 	if policy.GetMaximumAttempts() == 1 {
 		// One maximum attempt effectively disable retries. Validating the
 		// rest of the arguments is pointless
@@ -417,18 +429,61 @@ func ValidateRetryPolicy(policy *commonpb.RetryPolicy) error {
 	return nil
 }
 
+func GetDefaultRetryPolicyConfigOptions() map[string]interface{} {
+	return map[string]interface{}{
+		initialIntervalInSecondsConfigKey:   int(defaultInitialInterval.Seconds()),
+		maximumIntervalCoefficientConfigKey: defaultMaximumIntervalCoefficient,
+		backoffCoefficientConfigKey:         defaultBackoffCoefficient,
+		maximumAttemptsConfigKey:            defaultMaximumAttempts,
+	}
+}
+
+func FromConfigToDefaultRetrySettings(options map[string]interface{}) DefaultRetrySettings {
+	defaultSettings := DefaultRetrySettings{
+		InitialInterval:            defaultInitialInterval,
+		MaximumIntervalCoefficient: defaultMaximumIntervalCoefficient,
+		BackoffCoefficient:         defaultBackoffCoefficient,
+		MaximumAttempts:            defaultMaximumAttempts,
+	}
+
+	initialIntervalInSeconds, ok := options[initialIntervalInSecondsConfigKey]
+	if ok {
+		defaultSettings.InitialInterval = time.Duration(initialIntervalInSeconds.(int)) * time.Second
+	}
+
+	maximumIntervalCoefficient, ok := options[maximumIntervalCoefficientConfigKey]
+	if ok {
+		defaultSettings.MaximumIntervalCoefficient = maximumIntervalCoefficient.(float64)
+	}
+
+	backoffCoefficient, ok := options[backoffCoefficientConfigKey]
+	if ok {
+		defaultSettings.BackoffCoefficient = backoffCoefficient.(float64)
+	}
+
+	maximumAttempts, ok := options[maximumAttemptsConfigKey]
+	if ok {
+		defaultSettings.MaximumAttempts = int32(maximumAttempts.(int))
+	}
+
+	return defaultSettings
+}
+
 // CreateHistoryStartWorkflowRequest create a start workflow request for history
 func CreateHistoryStartWorkflowRequest(
 	namespaceID string,
 	startRequest *workflowservice.StartWorkflowExecutionRequest,
+	parentExecutionInfo *workflowspb.ParentExecutionInfo,
+	now time.Time,
 ) *historyservice.StartWorkflowExecutionRequest {
-	now := time.Now()
 	histRequest := &historyservice.StartWorkflowExecutionRequest{
 		NamespaceId:            namespaceID,
 		StartRequest:           startRequest,
 		ContinueAsNewInitiator: enumspb.CONTINUE_AS_NEW_INITIATOR_WORKFLOW,
 		Attempt:                1,
+		ParentExecutionInfo:    parentExecutionInfo,
 	}
+
 	if timestamp.DurationValue(startRequest.GetWorkflowExecutionTimeout()) > 0 {
 		deadline := now.Add(timestamp.DurationValue(startRequest.GetWorkflowExecutionTimeout()))
 		histRequest.WorkflowExecutionExpirationTime = timestamp.TimePtr(deadline.Round(time.Millisecond))

common/util.go

@@ -92,8 +92,8 @@ func TestValidateRetryPolicy(t *testing.T) {
 }
 
 func TestEnsureRetryPolicyDefaults(t *testing.T) {
-	defaultActivityRetrySettings := DefaultActivityRetrySettings{
-		InitialIntervalInSeconds:   1,
+	defaultRetrySettings := DefaultRetrySettings{
+		InitialInterval:            time.Second,
 		MaximumIntervalCoefficient: 100,
 		BackoffCoefficient:         2.0,
 		MaximumAttempts:            120,
@@ -181,8 +181,23 @@ func TestEnsureRetryPolicyDefaults(t *testing.T) {
 
 	for _, tt := range testCases {
 		t.Run(tt.name, func(t *testing.T) {
-			EnsureRetryPolicyDefaults(tt.input, defaultActivityRetrySettings)
+			EnsureRetryPolicyDefaults(tt.input, defaultRetrySettings)
 			assert.Equal(t, tt.want, tt.input)
 		})
 	}
 }
+
+func Test_FromConfigToRetryPolicy(t *testing.T) {
+	options := map[string]interface{}{
+		initialIntervalInSecondsConfigKey:   2,
+		maximumIntervalCoefficientConfigKey: 100.0,
+		backoffCoefficientConfigKey:         4.0,
+		maximumAttemptsConfigKey:            5,
+	}
+
+	defaultSettings := FromConfigToDefaultRetrySettings(options)
+	assert.Equal(t, 2*time.Second, defaultSettings.InitialInterval)
+	assert.Equal(t, 100.0, defaultSettings.MaximumIntervalCoefficient)
+	assert.Equal(t, 4.0, defaultSettings.BackoffCoefficient)
+	assert.Equal(t, int32(5), defaultSettings.MaximumAttempts)
+}

common/util_test.go

@@ -28,3 +28,9 @@ history.defaultActivityRetryPolicy:
     MaximumIntervalCoefficient: 100.0
     BackoffCoefficient: 2.0
     MaximumAttempts: 0
+history.defaultWorkflowRetryPolicy:
+- value:
+    InitialIntervalInSeconds: 1
+    MaximumIntervalCoefficient: 100.0
+    BackoffCoefficient: 2.0
+    MaximumAttempts: 0

config/dynamicconfig/development.yaml

@@ -615,7 +615,7 @@ func (s *integrationSuite) TestWorkflowRetry() {
 		WorkflowTaskTimeout: timestamp.DurationPtr(1 * time.Second),
 		Identity:            identity,
 		RetryPolicy: &commonpb.RetryPolicy{
-			InitialInterval:        &initialInterval,
+			// Intentionally test server-initialization of Initial Interval value (which should be 1 second)
 			MaximumAttempts:        int32(maximumAttempts),
 			MaximumInterval:        timestamp.DurationPtr(1 * time.Second),
 			NonRetryableErrorTypes: []string{"bad-bug"},

host/integration_test.go

@@ -97,6 +97,10 @@ type Config struct {
 	SearchAttributesSizeOfValueLimit  dynamicconfig.IntPropertyFnWithNamespaceFilter
 	SearchAttributesTotalSizeLimit    dynamicconfig.IntPropertyFnWithNamespaceFilter
 
+	// DefaultWorkflowRetryPolicy represents default values for unset fields on a Workflow's
+	// specified RetryPolicy
+	DefaultWorkflowRetryPolicy dynamicconfig.MapPropertyFnWithNamespaceFilter
+
 	// VisibilityArchival system protection
 	VisibilityArchivalQueryMaxPageSize dynamicconfig.IntPropertyFn
 
@@ -144,6 +148,7 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int, enableReadFro
 		SendRawWorkflowHistory:                 dc.GetBoolPropertyFnWithNamespaceFilter(dynamicconfig.SendRawWorkflowHistory, false),
 		EnableRPCReplication:                   dc.GetBoolProperty(dynamicconfig.FrontendEnableRPCReplication, false),
 		EnableCleanupReplicationTask:           dc.GetBoolProperty(dynamicconfig.FrontendEnableCleanupReplicationTask, true),
+		DefaultWorkflowRetryPolicy:             dc.GetMapPropertyFnWithNamespaceFilter(dynamicconfig.DefaultWorkflowRetryPolicy, common.GetDefaultRetryPolicyConfigOptions()),
 	}
 }