Scheduler Bugfix: getFutureActionTimes should ignore actions prior to update time, respect RemainingActions counter (#6122)

## What changed/why?
In #6028/#5381, our scheduler's run loop accounted for update requests
that occur between nominal and actual/jittered times. The run loop will
correctly skip over action times who are scheduled prior to the
schedule's update time (such as when a schedule is recalculated to a new
spec). This PR updates `getFutureActionTimes` (used as part of the
scheduler's describe and list operations) to account for action times
preceding the update time, as well as to properly reflect the number of
remaining actions in a schedule.

## How did you test it?
- Updated unit tests for both limited actions (`TestLimitedActions`) as
well as filtered update times (`TestUpdateNotRetroactive`).
`TestUpdateNotRetroactive` is the test that goes red when the [parallel
condition in the
scheduler](https://github.com/temporalio/temporal/blob/44a6fe777e1dc950dd278cb1b7f6ffc8b3ba4328/service/worker/scheduler/workflow.go#L655-L660)
is commented out, so it seemed the appropriate place to also test
`getFutureActionTimes` lines up with the fix in the run loop.
- Verified both updated tests only pass when new version
(`AccurateFutureActionTimes`) is set; verified no other tests break
(`make test-unit`)

## Potential risks
- We break a use case that relied on speculative future action times/a
constant number of action times returned

## Documentation
N/A

## Is hotfix candidate?
No

Author: lina-temporal

service/worker/scheduler/workflow.go

@@ -84,6 +84,8 @@ const (
 	CANAfterSignals = 7
 	// set LastProcessedTime to last action instead of now
 	UseLastAction = 8
+	// getFutureActionTimes accounts for UpdateTime and RemainingActions
+	AccurateFutureActionTimes = 9
 )
 
 const (
@@ -931,6 +933,11 @@ func (s *scheduler) processSignals() bool {
 	return scheduleChanged
 }
 
+// Returns up to `n` future action times.
+//
+// After workflow version `AccurateFutureActionTimes`, No more than the
+// schedule's `RemainingActions` will be returned. Future action times that
+// precede the schedule's UpdateTime are not included.
 func (s *scheduler) getFutureActionTimes(inWorkflowContext bool, n int) []*timestamppb.Timestamp {
 	// Note that `s` may be a `scheduler` created outside of a workflow context, used to
 	// compute list info at creation time or in a query. In that case inWorkflowContext will
@@ -950,6 +957,10 @@ func (s *scheduler) getFutureActionTimes(inWorkflowContext bool, n int) []*times
 		}
 	}
 
+	if s.hasMinVersion(AccurateFutureActionTimes) && s.Schedule.State.LimitedActions {
+		n = min(int(s.Schedule.State.RemainingActions), n)
+	}
+
 	if s.cspec == nil {
 		return nil
 	}
@@ -960,6 +971,12 @@ func (s *scheduler) getFutureActionTimes(inWorkflowContext bool, n int) []*times
 		if t1.IsZero() {
 			break
 		}
+
+		if s.hasMinVersion(AccurateFutureActionTimes) && s.Info.UpdateTime.AsTime().After(t1) {
+			// Skip action times whose nominal times are prior to the schedule's update time
+			continue
+		}
+
 		out = append(out, timestamppb.New(t1))
 	}
 	return out

service/worker/scheduler/workflow_test.go

@@ -1619,6 +1619,11 @@ func (s *workflowSuite) TestUpdate() {
 }
 
 func (s *workflowSuite) TestUpdateNotRetroactive() {
+	// TODO - remove when AccurateFutureActionTimes becomes the active version
+	prevTweakables := currentTweakablePolicies
+	currentTweakablePolicies.Version = AccurateFutureActionTimes
+	defer func() { currentTweakablePolicies = prevTweakables }()
+
 	s.runAcrossContinue(
 		[]workflowRun{
 			{
@@ -1656,6 +1661,16 @@ func (s *workflowSuite) TestUpdateNotRetroactive() {
 					})
 				},
 			},
+			// After the update above modifies the schedule, we should discard any newly
+			// scheduled times that are scheduled prior to the update time.
+			{
+				at: time.Date(2022, 6, 1, 1, 7, 12, 0, time.UTC),
+				f: func() {
+					desc := s.describe()
+					times := desc.Info.FutureActionTimes
+					s.True(times[0].AsTime().After(desc.Info.UpdateTime.AsTime()), "getFutureActionTimes returned an action preceding the update time after a schedule change")
+				},
+			},
 			{
 				at:         time.Date(2022, 6, 1, 1, 7, 55, 0, time.UTC),
 				finishTest: true,
@@ -1844,6 +1859,11 @@ func (s *workflowSuite) TestPauseUnpauseBetweenNominalAndJittered() {
 }
 
 func (s *workflowSuite) TestLimitedActions() {
+	// TODO - remove when AccurateFutureActionTimes becomes the active version
+	prevTweakables := currentTweakablePolicies
+	currentTweakablePolicies.Version = AccurateFutureActionTimes
+	defer func() { currentTweakablePolicies = prevTweakables }()
+
 	// written using low-level mocks so we can sleep forever
 
 	// limited to 2
@@ -1870,13 +1890,19 @@ func (s *workflowSuite) TestLimitedActions() {
 	})
 
 	s.env.RegisterDelayedCallback(func() {
-		s.Equal(int64(2), s.describe().Schedule.State.RemainingActions)
+		desc := s.describe()
+		s.Equal(int64(2), desc.Schedule.State.RemainingActions)
+		s.Equal(2, len(desc.Info.FutureActionTimes))
 	}, 1*time.Minute)
 	s.env.RegisterDelayedCallback(func() {
-		s.Equal(int64(1), s.describe().Schedule.State.RemainingActions)
+		desc := s.describe()
+		s.Equal(int64(1), desc.Schedule.State.RemainingActions)
+		s.Equal(1, len(desc.Info.FutureActionTimes))
 	}, 5*time.Minute)
 	s.env.RegisterDelayedCallback(func() {
-		s.Equal(int64(0), s.describe().Schedule.State.RemainingActions)
+		desc := s.describe()
+		s.Equal(int64(0), desc.Schedule.State.RemainingActions)
+		s.Equal(0, len(desc.Info.FutureActionTimes))
 		s.Equal(1, len(s.runningWorkflows()))
 	}, 7*time.Minute)
 	s.env.RegisterDelayedCallback(func() {