Calculate SchdeuleToCloseDeadline from current attempt instead of first attempt (#1022)

* Use GetScheduledTimestamp from current attempt instead of first attempt for SchdeuleToCloseDeadline

Author: anish531213

internal/activity.go

@@ -300,7 +300,7 @@ func WithActivityTask(
 	tracer opentracing.Tracer,
 ) context.Context {
 	var deadline time.Time
-	scheduled := time.Unix(0, task.GetScheduledTimestamp())
+	scheduled := time.Unix(0, task.GetScheduledTimestampOfThisAttempt())
 	started := time.Unix(0, task.GetStartedTimestamp())
 	scheduleToCloseTimeout := time.Duration(task.GetScheduleToCloseTimeoutSeconds()) * time.Second
 	startToCloseTimeout := time.Duration(task.GetStartToCloseTimeoutSeconds()) * time.Second

internal/internal_task_handlers_test.go

@@ -1300,12 +1300,13 @@ func (t *TaskHandlersTestSuite) TestActivityExecutionDeadline() {
 			WorkflowExecution: &s.WorkflowExecution{
 				WorkflowId: common.StringPtr("wID"),
 				RunId:      common.StringPtr("rID")},
-			ActivityType:                  &s.ActivityType{Name: common.StringPtr("test")},
-			ActivityId:                    common.StringPtr(uuid.New()),
-			ScheduledTimestamp:            common.Int64Ptr(d.ScheduleTS.UnixNano()),
-			ScheduleToCloseTimeoutSeconds: common.Int32Ptr(d.ScheduleDuration),
-			StartedTimestamp:              common.Int64Ptr(d.StartTS.UnixNano()),
-			StartToCloseTimeoutSeconds:    common.Int32Ptr(d.StartDuration),
+			ActivityType:                    &s.ActivityType{Name: common.StringPtr("test")},
+			ActivityId:                      common.StringPtr(uuid.New()),
+			ScheduledTimestamp:              common.Int64Ptr(d.ScheduleTS.UnixNano()),
+			ScheduledTimestampOfThisAttempt: common.Int64Ptr(d.ScheduleTS.UnixNano()),
+			ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(d.ScheduleDuration),
+			StartedTimestamp:                common.Int64Ptr(d.StartTS.UnixNano()),
+			StartToCloseTimeoutSeconds:      common.Int32Ptr(d.StartDuration),
 			WorkflowType: &s.WorkflowType{
 				Name: common.StringPtr("wType"),
 			},
@@ -1358,12 +1359,13 @@ func (t *TaskHandlersTestSuite) TestActivityExecutionWorkerStop() {
 		WorkflowExecution: &s.WorkflowExecution{
 			WorkflowId: common.StringPtr("wID"),
 			RunId:      common.StringPtr("rID")},
-		ActivityType:                  &s.ActivityType{Name: common.StringPtr("test")},
-		ActivityId:                    common.StringPtr(uuid.New()),
-		ScheduledTimestamp:            common.Int64Ptr(time.Now().UnixNano()),
-		ScheduleToCloseTimeoutSeconds: common.Int32Ptr(1),
-		StartedTimestamp:              common.Int64Ptr(time.Now().UnixNano()),
-		StartToCloseTimeoutSeconds:    common.Int32Ptr(1),
+		ActivityType:                    &s.ActivityType{Name: common.StringPtr("test")},
+		ActivityId:                      common.StringPtr(uuid.New()),
+		ScheduledTimestamp:              common.Int64Ptr(time.Now().UnixNano()),
+		ScheduledTimestampOfThisAttempt: common.Int64Ptr(time.Now().UnixNano()),
+		ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(1),
+		StartedTimestamp:                common.Int64Ptr(time.Now().UnixNano()),
+		StartToCloseTimeoutSeconds:      common.Int32Ptr(1),
 		WorkflowType: &s.WorkflowType{
 			Name: common.StringPtr("wType"),
 		},

internal/internal_workers_test.go

@@ -144,12 +144,13 @@ func (s *WorkersTestSuite) TestActivityWorkerStop() {
 		WorkflowExecution: &m.WorkflowExecution{
 			WorkflowId: common.StringPtr("wID"),
 			RunId:      common.StringPtr("rID")},
-		ActivityType:                  &m.ActivityType{Name: common.StringPtr("test")},
-		ActivityId:                    common.StringPtr(uuid.New()),
-		ScheduledTimestamp:            common.Int64Ptr(time.Now().UnixNano()),
-		ScheduleToCloseTimeoutSeconds: common.Int32Ptr(1),
-		StartedTimestamp:              common.Int64Ptr(time.Now().UnixNano()),
-		StartToCloseTimeoutSeconds:    common.Int32Ptr(1),
+		ActivityType:                    &m.ActivityType{Name: common.StringPtr("test")},
+		ActivityId:                      common.StringPtr(uuid.New()),
+		ScheduledTimestamp:              common.Int64Ptr(time.Now().UnixNano()),
+		ScheduledTimestampOfThisAttempt: common.Int64Ptr(time.Now().UnixNano()),
+		ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(1),
+		StartedTimestamp:                common.Int64Ptr(time.Now().UnixNano()),
+		StartToCloseTimeoutSeconds:      common.Int32Ptr(1),
 		WorkflowType: &m.WorkflowType{
 			Name: common.StringPtr("wType"),
 		},

internal/internal_workflow_testsuite.go

@@ -1575,15 +1575,16 @@ func newTestActivityTask(workflowID, runID, activityID, workflowTypeName, domain
 			WorkflowId: common.StringPtr(workflowID),
 			RunId:      common.StringPtr(runID),
 		},
-		ActivityId:                    common.StringPtr(activityID),
-		TaskToken:                     []byte(activityID), // use activityID as TaskToken so we can map TaskToken in heartbeat calls.
-		ActivityType:                  &shared.ActivityType{Name: common.StringPtr(params.ActivityType.Name)},
-		Input:                         params.Input,
-		ScheduledTimestamp:            common.Int64Ptr(time.Now().UnixNano()),
-		ScheduleToCloseTimeoutSeconds: common.Int32Ptr(params.ScheduleToCloseTimeoutSeconds),
-		StartedTimestamp:              common.Int64Ptr(time.Now().UnixNano()),
-		StartToCloseTimeoutSeconds:    common.Int32Ptr(params.StartToCloseTimeoutSeconds),
-		HeartbeatTimeoutSeconds:       common.Int32Ptr(params.HeartbeatTimeoutSeconds),
+		ActivityId:                      common.StringPtr(activityID),
+		TaskToken:                       []byte(activityID), // use activityID as TaskToken so we can map TaskToken in heartbeat calls.
+		ActivityType:                    &shared.ActivityType{Name: common.StringPtr(params.ActivityType.Name)},
+		Input:                           params.Input,
+		ScheduledTimestamp:              common.Int64Ptr(time.Now().UnixNano()),
+		ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(params.ScheduleToCloseTimeoutSeconds),
+		ScheduledTimestampOfThisAttempt: common.Int64Ptr(time.Now().UnixNano()),
+		StartedTimestamp:                common.Int64Ptr(time.Now().UnixNano()),
+		StartToCloseTimeoutSeconds:      common.Int32Ptr(params.StartToCloseTimeoutSeconds),
+		HeartbeatTimeoutSeconds:         common.Int32Ptr(params.HeartbeatTimeoutSeconds),
 		WorkflowType: &shared.WorkflowType{
 			Name: common.StringPtr(workflowTypeName),
 		},