Account for local activity run time in schedule sleep (#4079)

Author: dnr

service/worker/scheduler/workflow.go

@@ -76,8 +76,6 @@ const (
 	// query so it can be changed without breaking history.)
 	maxListMatchingTimesCount = 1000
 
-	invalidDuration time.Duration = -1
-
 	rateLimitedErrorType = "RateLimited"
 )
 
@@ -118,7 +116,8 @@ type (
 		SleepWhilePaused                  bool // If true, don't set timers while paused/out of actions
 		// MaxBufferSize limits the number of buffered starts. This also limits the number of
 		// workflows that can be backfilled at once (since they all have to fit in the buffer).
-		MaxBufferSize int
+		MaxBufferSize  int
+		AllowZeroSleep bool // Whether to allow a zero-length timer. Used for workflow compatibility.
 	}
 )
 
@@ -150,6 +149,7 @@ var (
 		IterationsBeforeContinueAsNew:     500,
 		SleepWhilePaused:                  true,
 		MaxBufferSize:                     1000,
+		AllowZeroSleep:                    true,
 	}
 
 	errUpdateConflict = errors.New("conflicting concurrent update")
@@ -204,7 +204,7 @@ func (s *scheduler) run() error {
 			s.logger.Warn("Time went backwards", "from", t1, "to", t2)
 			t2 = t1
 		}
-		nextSleep := s.processTimeRange(
+		nextWakeup := s.processTimeRange(
 			t1, t2,
 			// resolve this to the schedule's policy as late as possible
 			enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,
@@ -215,7 +215,7 @@ func (s *scheduler) run() error {
 		scheduleChanged := s.processSignals()
 		if scheduleChanged {
 			// need to calculate sleep again
-			nextSleep = s.processTimeRange(t2, t2, enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED, false)
+			nextWakeup = s.processTimeRange(t2, t2, enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED, false)
 		}
 		// try starting workflows in the buffer
 		//nolint:revive
@@ -226,7 +226,7 @@ func (s *scheduler) run() error {
 		// 1. requested time elapsed
 		// 2. we got a signal (update, request, refresh)
 		// 3. a workflow that we were watching finished
-		s.sleep(nextSleep)
+		s.sleep(nextWakeup)
 		s.updateTweakables()
 	}
 
@@ -333,11 +333,11 @@ func (s *scheduler) processTimeRange(
 	t1, t2 time.Time,
 	overlapPolicy enumspb.ScheduleOverlapPolicy,
 	manual bool,
-) time.Duration {
+) time.Time {
 	s.logger.Debug("processTimeRange", "t1", t1, "t2", t2, "overlap-policy", overlapPolicy, "manual", manual)
 
 	if s.cspec == nil {
-		return invalidDuration
+		return time.Time{}
 	}
 
 	catchupWindow := s.getCatchupWindow()
@@ -350,10 +350,8 @@ func (s *scheduler) processTimeRange(
 			return s.cspec.getNextTime(t1)
 		}).Get(&next))
 		t1 = next.Next
-		if t1.IsZero() {
-			return invalidDuration
-		} else if t1.After(t2) {
-			return t1.Sub(t2)
+		if t1.IsZero() || t1.After(t2) {
+			return t1
 		}
 		if !manual && t2.Sub(t1) > catchupWindow {
 			s.logger.Warn("Schedule missed catchup window", "now", t2, "time", t1)
@@ -395,7 +393,7 @@ func (s *scheduler) canTakeScheduledAction(manual, decrement bool) bool {
 	return false
 }
 
-func (s *scheduler) sleep(nextSleep time.Duration) {
+func (s *scheduler) sleep(nextWakeup time.Time) {
 	sel := workflow.NewSelector(s.ctx)
 
 	upCh := workflow.GetSignalChannel(s.ctx, SignalNameUpdate)
@@ -413,19 +411,27 @@ func (s *scheduler) sleep(nextSleep time.Duration) {
 
 	// if we're paused or out of actions, we don't need to wake up until we get an update
 	if s.tweakables.SleepWhilePaused && !s.canTakeScheduledAction(false, false) {
-		nextSleep = invalidDuration
+		nextWakeup = time.Time{}
 	}
 
-	if nextSleep != invalidDuration {
-		tmr := workflow.NewTimer(s.ctx, nextSleep)
+	if !nextWakeup.IsZero() {
+		sleepTime := nextWakeup.Sub(s.now())
+		// A previous version of this workflow passed around sleep duration instead of wakeup time,
+		// which means it always set a timer even in cases where sleepTime comes out negative. For
+		// compatibility, we have to continue setting a positive timer in those cases. The value
+		// doesn't have to match, though.
+		if !s.tweakables.AllowZeroSleep && sleepTime <= 0 {
+			sleepTime = time.Second
+		}
+		tmr := workflow.NewTimer(s.ctx, sleepTime)
 		sel.AddFuture(tmr, func(_ workflow.Future) {})
 	}
 
 	if s.watchingFuture != nil {
 		sel.AddFuture(s.watchingFuture, s.wfWatcherReturned)
 	}
 
-	s.logger.Debug("sleeping", "next-sleep", nextSleep, "watching", s.watchingFuture != nil)
+	s.logger.Debug("sleeping", "next-wakeup", nextWakeup, "watching", s.watchingFuture != nil)
 	sel.Select(s.ctx)
 	for sel.HasPending() {
 		sel.Select(s.ctx)