feat: restart delay period

The restart delay period should allow the attempt to be reset if
the worker is restarted after that time. This fixes the issue where
we might see the attempt climbing even though the restarts over a
long period of time.

This is backwards compatible with the v4 version, as the restart
delay is a constant, so the attempt is disguarded.

Author: SimonRichardson

runner.go

@@ -26,12 +26,12 @@ const (
 
 // DelayFunc is the type alias of the function that returns the delay between
 // worker restarts. It is called with the number of attempts that have been made
-// to start the worker, the time the worker was last restarted, and the error
-// that caused the worker to exit.
+// since the last window period. The lastErr argument is the error that caused
+// the worker to exit. Depending on the error, the delay can be adjusted.
 //
 // The worker id is not passed to the function because we shouldn't to prevent
 // the function from being able to make decisions based on the worker id.
-type DelayFunc = func(attempts int, lastRestartedTime time.Time, lastErr error) time.Duration
+type DelayFunc = func(attempts int, lastErr error) time.Duration
 
 // StartFunc is the type alias of the function that creates a worker.
 type StartFunc = func(context.Context) (Worker, error)
@@ -191,6 +191,10 @@ type RunnerParams struct {
 	// If this is nil, DefaultRestartDelay will be used.
 	RestartDelay DelayFunc
 
+	// RestartDelayPeriod is the length of time that the restart delay
+	// is calculated over.
+	RestartDelayPeriod time.Duration
+
 	// Clock is used for timekeeping. If it's nil, clock.WallClock
 	// will be used.
 	Clock Clock
@@ -239,10 +243,13 @@ func NewRunner(p RunnerParams) (*Runner, error) {
 		}
 	}
 	if p.RestartDelay == nil {
-		p.RestartDelay = func(attempts int, lastRestartedTime time.Time, lastErr error) time.Duration {
+		p.RestartDelay = func(attempts int, lastErr error) time.Duration {
 			return DefaultRestartDelay
 		}
 	}
+	if p.RestartDelayPeriod == 0 {
+		p.RestartDelayPeriod = 10 * time.Minute
+	}
 	if p.Clock == nil {
 		p.Clock = clock.WallClock
 	}
@@ -578,12 +585,20 @@ func (runner *Runner) workerDone(info doneInfo) {
 		return
 	}
 
-	// The worker has exited with a non-fatal error.
-	// We'll restart it after a delay, increment the attempts, so it's
-	// possible to track how many times the worker has been restarted.
-	delay := workerInfo.restartDelay(workerInfo.attempts, workerInfo.restarted, info.err)
-	workerInfo.attempts++
-	workerInfo.restarted = params.Clock.Now().UTC()
+	// The worker has exited with a non-fatal error. We'll restart it after a
+	// delay, increment the attempts, so it's possible to track how many times
+	// the worker has been restarted over the restart delay period.
+	now := params.Clock.Now().UTC()
+
+	// If the worker has been restarted after the restart delay period, reset
+	// the attempts counter. Otherwise, increment the attempts counter.
+	if now.After(workerInfo.restarted.Add(params.RestartDelayPeriod)) {
+		workerInfo.attempts = 0
+	} else {
+		workerInfo.attempts++
+	}
+	delay := workerInfo.restartDelay(workerInfo.attempts, info.err)
+	workerInfo.restarted = now
 
 	// Kick off the worker again taking into account the delay.
 	pprof.Do(runner.tomb.Context(context.Background()), workerInfo.labels, func(ctx context.Context) {

runner_test.go

@@ -370,7 +370,7 @@ func (*RunnerSuite) TestOneWorkerRestartDelayAfterFailedAttempts(c *gc.C) {
 	runner, err := worker.NewRunner(worker.RunnerParams{
 		Name:    "test",
 		IsFatal: noneFatal,
-		RestartDelay: func(attempts int, lastRestart time.Time, restartErr error) time.Duration {
+		RestartDelay: func(attempts int, restartErr error) time.Duration {
 			received.Add(int64(attempts))
 
 			go func(restartErr error) {
@@ -411,31 +411,22 @@ func (*RunnerSuite) TestOneWorkerRestartDelayAfterFailedAttempts(c *gc.C) {
 	}
 	c.Assert(worker.Stop(runner), gc.IsNil)
 
-	c.Check(received.Load(), gc.Equals, int64(1))
+	c.Check(received.Load(), gc.Equals, int64(0))
 }
 
-func (*RunnerSuite) TestOneWorkerRestartDelayAfterFailedAttemptsLastRestart(c *gc.C) {
+func (*RunnerSuite) TestOneWorkerRestartDelayAfterFailedAttemptsLargeDelayPeriod(c *gc.C) {
 	var received atomic.Int64
 
-	restarts := make(chan time.Time)
-
 	const delay = 100 * time.Millisecond
 	runner, err := worker.NewRunner(worker.RunnerParams{
 		Name:    "test",
 		IsFatal: noneFatal,
-		RestartDelay: func(attempts int, lastRestart time.Time, restartErr error) time.Duration {
+		RestartDelay: func(attempts int, restartErr error) time.Duration {
 			received.Add(int64(attempts))
 
-			go func(lastRestart time.Time) {
-				select {
-				case restarts <- lastRestart:
-				case <-time.After(shortWait):
-					c.Fatalf("timed out sending restart error")
-				}
-			}(lastRestart)
-
 			return delay
 		},
+		RestartDelayPeriod: delay * 2,
 	})
 	c.Assert(err, jc.ErrorIsNil)
 
@@ -448,29 +439,61 @@ func (*RunnerSuite) TestOneWorkerRestartDelayAfterFailedAttemptsLastRestart(c *g
 	starter.die <- errors.New("fatal error")
 	starter.assertStarted(c, false)
 
-	select {
-	case restart := <-restarts:
-		// Initial restart time will be empty, as no restart has happened.
-		c.Check(restart, gc.Equals, time.Time{})
-	case <-time.After(shortWait):
-		c.Fatalf("timed out waiting for restart error ")
-	}
+	// First attempt should always be zero.
+	c.Check(received.Load(), gc.Equals, int64(0))
 
 	starter.assertStarted(c, true)
 	starter.die <- errors.New("fatal error")
 	starter.assertStarted(c, false)
 
-	select {
-	case restart := <-restarts:
-		c.Check(restart.IsZero(), jc.IsFalse)
-	case <-time.After(shortWait):
-		c.Fatalf("timed out waiting for restart error ")
-	}
+	starter.assertStarted(c, true)
+
+	c.Assert(worker.Stop(runner), gc.IsNil)
+
+	// Second attempt with a large delay period should not reset the attempt
+	// counter.
+	c.Check(received.Load(), gc.Equals, int64(1))
+}
+
+func (*RunnerSuite) TestOneWorkerRestartDelayAfterFailedAttemptsSmallDelayPeriod(c *gc.C) {
+	var received atomic.Int64
+
+	const delay = 100 * time.Millisecond
+	runner, err := worker.NewRunner(worker.RunnerParams{
+		Name:    "test",
+		IsFatal: noneFatal,
+		RestartDelay: func(attempts int, restartErr error) time.Duration {
+			received.Add(int64(attempts))
+
+			return delay
+		},
+		RestartDelayPeriod: delay / 2,
+	})
+	c.Assert(err, jc.ErrorIsNil)
+
+	starter := newTestWorkerStarter()
+
+	err = runner.StartWorker(context.Background(), "id", starter.start)
+	c.Assert(err, jc.ErrorIsNil)
+
+	starter.assertStarted(c, true)
+	starter.die <- errors.New("fatal error")
+	starter.assertStarted(c, false)
+
+	// First attempt should always be zero.
+	c.Check(received.Load(), gc.Equals, int64(0))
+
+	starter.assertStarted(c, true)
+	starter.die <- errors.New("fatal error")
+	starter.assertStarted(c, false)
 
 	starter.assertStarted(c, true)
 
 	c.Assert(worker.Stop(runner), gc.IsNil)
-	c.Check(received.Load(), gc.Equals, int64(3))
+
+	// Second attempt with a small delay period should reset the attempt
+	// counter.
+	c.Check(received.Load(), gc.Equals, int64(0))
 }
 
 func (*RunnerSuite) TestOneWorkerShouldNotRestart(c *gc.C) {
@@ -1193,7 +1216,7 @@ func (checker *containsChecker) Check(params []interface{}, names []string) (boo
 }
 
 func constDelay(d time.Duration) worker.DelayFunc {
-	return func(attempts int, lastRestartTime time.Time, lastError error) time.Duration {
+	return func(attempts int, lastError error) time.Duration {
 		return d
 	}
 }