Add hwm

Author: dandavison

chasm/lib/activity/activity_tasks.go

@@ -9,6 +9,7 @@ import (
 	"go.temporal.io/server/common/resource"
 	"go.temporal.io/server/common/util"
 	"go.uber.org/fx"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type activityDispatchTaskExecutorOptions struct {
@@ -159,41 +160,43 @@ func newHeartbeatTimeoutTaskExecutor() *heartbeatTimeoutTaskExecutor {
 func (e *heartbeatTimeoutTaskExecutor) Validate(
 	ctx chasm.Context,
 	activity *Activity,
-	_ chasm.TaskAttributes,
+	taskAttrs chasm.TaskAttributes,
 	task *activitypb.HeartbeatTimeoutTask,
 ) (bool, error) {
 	validStatus := activity.Status == activitypb.ACTIVITY_EXECUTION_STATUS_STARTED ||
 		activity.Status == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED
-	return validStatus && activity.LastAttempt.Get(ctx).GetCount() == task.Attempt, nil
+	if !validStatus || activity.LastAttempt.Get(ctx).GetCount() != task.Attempt {
+		return false, nil
+	}
+	// High-water-mark: reject tasks that have already been executed.
+	hwm := activity.GetLastHeartbeatTaskScheduledTime().AsTime()
+	return taskAttrs.ScheduledTime.After(hwm), nil
 }
 
 // Execute executes a HeartbeatTimeoutTask.
 func (e *heartbeatTimeoutTaskExecutor) Execute(
 	ctx chasm.MutableContext,
 	activity *Activity,
-	_ chasm.TaskAttributes,
+	taskAttrs chasm.TaskAttributes,
 	_ *activitypb.HeartbeatTimeoutTask,
 ) error {
-	// Let T = user-configured heartbeat timeout and let hb_i be the time of the ith user-submitted
-	// heartbeat request. (hb_0 = 0 since we always start a timer task when an attempt starts).
-
 	// There are two concurrent processes:
-	// 1. A worker is sending heartbeats at times hb_i.
+	// 1. A worker is sending heartbeats.
 	// 2. This task is being executed at (shortly after) certain scheduled times.
 
 	// Each time we execute this function, our task is to look back into the past and determine
-	// whether more than T has elapsed since the last heartbeat. If it has, we fail the attempt (and
-	// decide between retrying or failing the activity). If it has not, then we schedule a new timer
-	// task to execute this function at the new deadline, i.e. lastHeartbeatTime+HeartbeatTimeout.
-	//
-	// Task validation has established that an attempt is currently in progress and that it is the
-	// attempt for which this heartbeat timer was originally set.
+	// whether more than (user-configured heartbeat timeout) has elapsed since the last heartbeat.
+	// If it has, we fail the attempt (and decide between retrying or failing the activity). If it
+	// has not, then we schedule a new timer task to execute this function at the new deadline.
+
+	// Update high-water-mark so this task is invalidated after execution.
+	activity.LastHeartbeatTaskScheduledTime = timestamppb.New(taskAttrs.ScheduledTime)
 
 	attempt := activity.LastAttempt.Get(ctx)
-	lastHb, _ := activity.LastHeartbeat.TryGet(ctx)
 	hbTimeout := activity.GetHeartbeatTimeout().AsDuration()
 	attemptStartTime := attempt.GetStartedTime().AsTime()
-	lastHbTime := lastHb.GetRecordedTime().AsTime() // could be from a previous attempt or could be zero
+	heartbeat, _ := activity.LastHeartbeat.TryGet(ctx)
+	lastHbTime := heartbeat.GetRecordedTime().AsTime() // could be from a previous attempt or could be zero
 	// No heartbeats in the attempt so far is equivalent to a heartbeat having been sent at attempt
 	// start time.
 	hbDeadline := util.MaxTime(lastHbTime, attemptStartTime).Add(hbTimeout)

chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go

@@ -80,22 +80,25 @@ message ActivityState {
     google.protobuf.Duration start_to_close_timeout = 5;
     // Maximum permitted time between successful worker heartbeats.
     google.protobuf.Duration heartbeat_timeout = 6;
+    // High-water-mark for heartbeat timeout task validation. Updated when a heartbeat
+    // timeout task executes. Tasks with ScheduledTime <= this value are stale.
+    google.protobuf.Timestamp last_heartbeat_task_scheduled_time = 7;
     // The retry policy for the activity. Will never exceed `schedule_to_close_timeout`.
-    temporal.api.common.v1.RetryPolicy retry_policy = 7;
+    temporal.api.common.v1.RetryPolicy retry_policy = 8;
 
     // All of the possible activity statuses (covers both the public ActivityExecutionStatus and PendingActivityState).
     // TODO: consider moving this into ActivityAttemptState and renaming that message. This could save mutating two
     // components on each attempt transition.
-    ActivityExecutionStatus status = 8;
+    ActivityExecutionStatus status = 9;
 
     // Time the activity was originally scheduled via a StartActivityExecution request.
-    google.protobuf.Timestamp schedule_time = 9;
+    google.protobuf.Timestamp schedule_time = 10;
 
     // Priority metadata.
-    temporal.api.common.v1.Priority priority = 10;
+    temporal.api.common.v1.Priority priority = 11;
 
     // Set if activity cancelation was requested.
-    ActivityCancelState cancel_state = 11;
+    ActivityCancelState cancel_state = 12;
 }
 
 message ActivityCancelState {

chasm/lib/activity/proto/v1/activity_state.proto

@@ -13,7 +13,7 @@ import "google/protobuf/timestamp.proto";
 
 // CHASM scheduler top-level state.
 message SchedulerState {
-    // Scheduler request parameters and metadata. 
+    // Scheduler request parameters and metadata.
     temporal.api.schedule.v1.Schedule schedule = 2;
     temporal.api.schedule.v1.ScheduleInfo info = 3;