Support CLI Delay Start (cadence-workflow#4097)

* add delayed start support

* Address comments

* Add delay start support to cli

* Add integration test

common/types/mapper/thrift/shared.go

@@ -5039,6 +5039,7 @@ func FromStartWorkflowExecutionRequest(t *types.StartWorkflowExecutionRequest) *
 		Memo:                                FromMemo(t.Memo),
 		SearchAttributes:                    FromSearchAttributes(t.SearchAttributes),
 		Header:                              FromHeader(t.Header),
+		DelayStartSeconds:                   t.DelayStartSeconds,
 	}
 }
 
@@ -5063,6 +5064,7 @@ func ToStartWorkflowExecutionRequest(t *shared.StartWorkflowExecutionRequest) *t
 		Memo:                                ToMemo(t.Memo),
 		SearchAttributes:                    ToSearchAttributes(t.SearchAttributes),
 		Header:                              ToHeader(t.Header),
+		DelayStartSeconds:                   t.DelayStartSeconds,
 	}
 }
 

common/types/shared.go

@@ -7518,6 +7518,7 @@ type SignalWithStartWorkflowExecutionRequest struct {
 	Memo                                *Memo                  `json:"memo,omitempty"`
 	SearchAttributes                    *SearchAttributes      `json:"searchAttributes,omitempty"`
 	Header                              *Header                `json:"header,omitempty"`
+	DelayStartSeconds                   *int32                 `json:"delayStartSeconds,omitempty"`
 }
 
 // GetDomain is an internal getter (TBD...)
@@ -7576,6 +7577,14 @@ func (v *SignalWithStartWorkflowExecutionRequest) GetTaskStartToCloseTimeoutSeco
 	return
 }
 
+// GetDelayStartSeconds is an internal getter (TBD...)
+func (v *SignalWithStartWorkflowExecutionRequest) GetDelayStartSeconds() (o int32) {
+	if v != nil && v.DelayStartSeconds != nil {
+		return *v.DelayStartSeconds
+	}
+	return
+}
+
 // GetIdentity is an internal getter (TBD...)
 func (v *SignalWithStartWorkflowExecutionRequest) GetIdentity() (o string) {
 	if v != nil {
@@ -7955,6 +7964,7 @@ type StartChildWorkflowExecutionInitiatedEventAttributes struct {
 	Header                              *Header                `json:"header,omitempty"`
 	Memo                                *Memo                  `json:"memo,omitempty"`
 	SearchAttributes                    *SearchAttributes      `json:"searchAttributes,omitempty"`
+	DelayStartSeconds                   *int32                 `json:"delayStartSeconds,omitempty"`
 }
 
 // GetDomain is an internal getter (TBD...)
@@ -8013,6 +8023,14 @@ func (v *StartChildWorkflowExecutionInitiatedEventAttributes) GetTaskStartToClos
 	return
 }
 
+// GetDelayStartSeconds is an internal getter (TBD...)
+func (v *StartChildWorkflowExecutionInitiatedEventAttributes) GetDelayStartSeconds() (o int32) {
+	if v != nil && v.DelayStartSeconds != nil {
+		return *v.DelayStartSeconds
+	}
+	return
+}
+
 // GetParentClosePolicy is an internal getter (TBD...)
 func (v *StartChildWorkflowExecutionInitiatedEventAttributes) GetParentClosePolicy() (o ParentClosePolicy) {
 	if v != nil && v.ParentClosePolicy != nil {
@@ -8146,6 +8164,7 @@ type StartWorkflowExecutionRequest struct {
 	Memo                                *Memo                  `json:"memo,omitempty"`
 	SearchAttributes                    *SearchAttributes      `json:"searchAttributes,omitempty"`
 	Header                              *Header                `json:"header,omitempty"`
+	DelayStartSeconds                   *int32                 `json:"delayStartSeconds,omitempty"`
 }
 
 // GetDomain is an internal getter (TBD...)
@@ -8204,6 +8223,14 @@ func (v *StartWorkflowExecutionRequest) GetTaskStartToCloseTimeoutSeconds() (o i
 	return
 }
 
+// GetDelayStartSeconds is an internal getter (TBD...)
+func (v *StartWorkflowExecutionRequest) GetDelayStartSeconds() (o int32) {
+	if v != nil && v.DelayStartSeconds != nil {
+		return *v.DelayStartSeconds
+	}
+	return
+}
+
 // GetIdentity is an internal getter (TBD...)
 func (v *StartWorkflowExecutionRequest) GetIdentity() (o string) {
 	if v != nil {

common/util.go

@@ -103,6 +103,7 @@ var (
 	ErrContextTimeoutTooShort = &types.BadRequestError{Message: "Context timeout is too short."}
 	// ErrContextTimeoutNotSet is error for not setting a context timeout when calling a long poll API
 	ErrContextTimeoutNotSet = &types.BadRequestError{Message: "Context timeout is not set."}
+	ErrDelayStartSeconds    = &types.BadRequestError{Message: "Conflicting inputs: both DelayStartSeconds and Cron schedule is set"}
 )
 
 // AwaitWaitGroup calls Wait on the given wait
@@ -421,12 +422,22 @@ func CreateHistoryStartWorkflowRequest(
 	domainID string,
 	startRequest *types.StartWorkflowExecutionRequest,
 	now time.Time,
-) *types.HistoryStartWorkflowExecutionRequest {
+) (*types.HistoryStartWorkflowExecutionRequest, error) {
 	histRequest := &types.HistoryStartWorkflowExecutionRequest{
 		DomainUUID:   domainID,
 		StartRequest: startRequest,
 	}
-	firstDecisionTaskBackoffSeconds := backoff.GetBackoffForNextScheduleInSeconds(startRequest.GetCronSchedule(), now, now)
+
+	firstDecisionTaskBackoffSeconds := backoff.GetBackoffForNextScheduleInSeconds(
+		startRequest.GetCronSchedule(), now, now)
+	delayStartSeconds := startRequest.GetDelayStartSeconds()
+	if delayStartSeconds > 0 && firstDecisionTaskBackoffSeconds > 0 {
+		return nil, ErrDelayStartSeconds
+	}
+	if delayStartSeconds > 0 {
+		firstDecisionTaskBackoffSeconds = delayStartSeconds
+	}
+
 	histRequest.FirstDecisionTaskBackoffSeconds = Int32Ptr(firstDecisionTaskBackoffSeconds)
 
 	if startRequest.RetryPolicy != nil && startRequest.RetryPolicy.GetExpirationIntervalInSeconds() > 0 {
@@ -436,7 +447,7 @@ func CreateHistoryStartWorkflowRequest(
 		histRequest.ExpirationTimestamp = Int64Ptr(deadline.Round(time.Millisecond).UnixNano())
 	}
 
-	return histRequest
+	return histRequest, nil
 }
 
 // CheckEventBlobSizeLimit checks if a blob data exceeds limits. It logs a warning if it exceeds warnLimit,

common/util_test.go

@@ -97,13 +97,33 @@ func TestCreateHistoryStartWorkflowRequest_ExpirationTimeWithCron(t *testing.T)
 		CronSchedule: "@every 300s",
 	}
 	now := time.Now()
-	startRequest := CreateHistoryStartWorkflowRequest(domainID, request, now)
+	startRequest, _ := CreateHistoryStartWorkflowRequest(domainID, request, now)
+	require.NotNil(t, startRequest)
 
 	expirationTime := startRequest.GetExpirationTimestamp()
 	require.NotNil(t, expirationTime)
 	require.True(t, time.Unix(0, expirationTime).Sub(now) > 60*time.Second)
 }
 
+func TestCreateHistoryStartWorkflowRequest_DelayStart(t *testing.T) {
+	domainID := uuid.New()
+	request := &types.StartWorkflowExecutionRequest{
+		RetryPolicy: &types.RetryPolicy{
+			InitialIntervalInSeconds:    60,
+			ExpirationIntervalInSeconds: 60,
+		},
+		DelayStartSeconds: Int32Ptr(100),
+	}
+	now := time.Now()
+	startRequest, _ := CreateHistoryStartWorkflowRequest(domainID, request, now)
+	require.NotNil(t, startRequest)
+
+	expirationTime := startRequest.GetExpirationTimestamp()
+	require.NotNil(t, expirationTime)
+	require.True(t, time.Unix(0, expirationTime).Sub(now) > (100+60)*time.Second)
+	require.True(t, time.Unix(0, expirationTime).Sub(now) < (100+65)*time.Second)
+}
+
 func TestCreateHistoryStartWorkflowRequest_ExpirationTimeWithoutCron(t *testing.T) {
 	domainID := uuid.New()
 	request := &types.StartWorkflowExecutionRequest{
@@ -113,7 +133,8 @@ func TestCreateHistoryStartWorkflowRequest_ExpirationTimeWithoutCron(t *testing.
 		},
 	}
 	now := time.Now()
-	startRequest := CreateHistoryStartWorkflowRequest(domainID, request, now)
+	startRequest, _ := CreateHistoryStartWorkflowRequest(domainID, request, now)
+	require.NotNil(t, startRequest)
 
 	expirationTime := startRequest.GetExpirationTimestamp()
 	require.NotNil(t, expirationTime)

go.mod

@@ -58,7 +58,7 @@ require (
 	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
 	go.opencensus.io v0.22.5 // indirect
 	go.uber.org/atomic v1.5.1
-	go.uber.org/cadence v0.16.0
+	go.uber.org/cadence v0.16.1-0.20210330222409-fcc703e94e2f
 	go.uber.org/fx v1.10.0
 	go.uber.org/multierr v1.4.0
 	go.uber.org/thriftrw v1.25.0