Fixes instances where an activity heartbeat timeout would get "lost" (#771)

If, for any reason, the cached in-memory mutable state of a Workflow were to be reset (e.g. shard loading or transient persistent error), any in-flight activities w/heartbeats for that Workflow would lose their associated heartbeat timer tasks. In other words, it could result in an activity whose heartbeat has long expired, but which the server doesn't trying to re-schedule.

The fix is to populate the pendingActivityTimerHeartbeats map when loading the mutable state from the persistence layer if we have indicated that a heartbeat task has been created. This ensures that we don't have a pending activity without an in-memory record of a heartbeat when processing timer tasks.

Author: mastermanu

service/history/mutableStateBuilder.go

@@ -279,6 +279,11 @@ func (e *mutableStateBuilder) Load(
 	e.pendingActivityInfoIDs = state.ActivityInfos
 	for _, activityInfo := range state.ActivityInfos {
 		e.pendingActivityIDToEventID[activityInfo.ActivityId] = activityInfo.ScheduleId
+		if (activityInfo.TimerTaskStatus & timerTaskStatusCreatedHeartbeat) > 0 {
+			// Sets last pending timer heartbeat to year 2000.
+			// This ensures at least one heartbeat task will be processed for the pending activity.
+			e.pendingActivityTimerHeartbeats[activityInfo.ScheduleId] = time.Unix(946684800, 0)
+		}
 	}
 	e.pendingTimerInfoIDs = state.TimerInfos
 	for _, timerInfo := range state.TimerInfos {

service/history/timerQueueActiveTaskExecutor_test.go

@@ -824,7 +824,6 @@ func (s *timerQueueActiveTaskExecutorSuite) TestProcessActivityTimeout_Heartbeat
 	s.True(modified)
 	task := mutableState.insertTimerTasks[0]
 	s.Equal(enumspb.TIMEOUT_TYPE_HEARTBEAT, task.(*persistence.ActivityTimeoutTask).TimeoutType)
-	protoTaskTime := task.(*persistence.ActivityTimeoutTask).GetVisibilityTimestamp().Add(-time.Second)
 	s.NoError(err)
 	timerTask := &persistenceblobs.TimerTaskInfo{
 		ScheduleAttempt: 1,
@@ -835,7 +834,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestProcessActivityTimeout_Heartbeat
 		TaskId:          int64(100),
 		TaskType:        enumsspb.TASK_TYPE_ACTIVITY_TIMEOUT,
 		TimeoutType:     enumspb.TIMEOUT_TYPE_HEARTBEAT,
-		VisibilityTime:  &protoTaskTime,
+		VisibilityTime:  &time.Time{},
 		EventId:         scheduledEvent.GetEventId(),
 	}
 

service/history/timerQueueStandbyTaskExecutor_test.go

@@ -600,7 +600,6 @@ func (s *timerQueueStandbyTaskExecutorSuite) TestProcessActivityTimeout_Heartbea
 	s.True(modified)
 	task := mutableState.insertTimerTasks[0]
 	s.Equal(enumspb.TIMEOUT_TYPE_HEARTBEAT, task.(*persistence.ActivityTimeoutTask).TimeoutType)
-	protoTaskTime := task.(*persistence.ActivityTimeoutTask).GetVisibilityTimestamp().Add(-time.Second)
 	s.NoError(err)
 	timerTask := &persistenceblobs.TimerTaskInfo{
 		ScheduleAttempt: 1,
@@ -611,7 +610,7 @@ func (s *timerQueueStandbyTaskExecutorSuite) TestProcessActivityTimeout_Heartbea
 		TaskId:          int64(100),
 		TaskType:        enumsspb.TASK_TYPE_ACTIVITY_TIMEOUT,
 		TimeoutType:     enumspb.TIMEOUT_TYPE_HEARTBEAT,
-		VisibilityTime:  &protoTaskTime,
+		VisibilityTime:  &time.Time{},
 		EventId:         scheduledEvent.GetEventId(),
 	}