fix: restart spooler when broker terminates subscription (#600)

Fixes #577 (again). The original fix, #593, left out the agent spooler
system, which is responsible for relaying run events to the agent. So
when its subscription was terminated, it didn't restart itself.

This PR changes that: the spooler will now restart itself whenever its
subscription is terminated.

internal/agent/spooler.go

@@ -57,14 +57,14 @@ func newSpooler(app client, logger logr.Logger, cfg Config) *spoolerDaemon {
 	}
 }
 
-// start starts the spooler
+// start the spooler
 func (s *spoolerDaemon) start(ctx context.Context) error {
 	op := func() error {
 		return s.reinitialize(ctx)
 	}
 	policy := backoff.WithContext(backoff.NewExponentialBackOff(), ctx)
 	return backoff.RetryNotify(op, policy, func(err error, next time.Duration) {
-		s.Error(fmt.Errorf("%w: reconnecting in %s", err, next), "stream update")
+		s.Error(err, "restarting spooler", "backoff", next)
 	})
 }
 
@@ -114,7 +114,7 @@ func (s *spoolerDaemon) reinitialize(ctx context.Context) error {
 			return err
 		}
 	}
-	return nil
+	return pubsub.ErrSubscriptionTerminated
 }
 
 func (s *spoolerDaemon) handleEvent(ev pubsub.Event) error {

internal/agent/spooler_test.go

@@ -44,7 +44,7 @@ func TestSpooler(t *testing.T) {
 	assert.Equal(t, cancelation{Run: run4, Forceful: false}, <-spooler.getCancelation())
 	assert.Equal(t, cancelation{Run: run5, Forceful: true}, <-spooler.getCancelation())
 	cancel()
-	assert.NoError(t, <-errch)
+	assert.Equal(t, pubsub.ErrSubscriptionTerminated, <-errch)
 }
 
 func TestSpooler_handleEvent(t *testing.T) {

internal/daemon/daemon.go

@@ -371,24 +371,21 @@ func (d *Daemon) Start(ctx context.Context, started chan struct{}) error {
 
 	subsystems := []*Subsystem{
 		{
-			Name:           "broker",
-			BackoffRestart: true,
-			Logger:         d.Logger,
-			System:         d.Broker,
+			Name:   "broker",
+			Logger: d.Logger,
+			System: d.Broker,
 		},
 		{
-			Name:           "proxy",
-			BackoffRestart: true,
-			Logger:         d.Logger,
-			System:         d.LogsService,
+			Name:   "proxy",
+			Logger: d.Logger,
+			System: d.LogsService,
 		},
 		{
-			Name:           "reporter",
-			BackoffRestart: true,
-			Logger:         d.Logger,
-			Exclusive:      true,
-			DB:             d.DB,
-			LockID:         internal.Int64(run.ReporterLockID),
+			Name:      "reporter",
+			Logger:    d.Logger,
+			Exclusive: true,
+			DB:        d.DB,
+			LockID:    internal.Int64(run.ReporterLockID),
 			System: &run.Reporter{
 				Logger:                      d.Logger.WithValues("component", "reporter"),
 				VCSProviderService:          d.VCSProviderService,
@@ -399,9 +396,8 @@ func (d *Daemon) Start(ctx context.Context, started chan struct{}) error {
 			},
 		},
 		{
-			Name:           "webhook purger",
-			BackoffRestart: true,
-			Logger:         d.Logger,
+			Name:   "webhook purger",
+			Logger: d.Logger,
 			System: &repo.Purger{
 				Logger:     d.Logger.WithValues("component", "purger"),
 				Subscriber: d.Broker,
@@ -410,12 +406,11 @@ func (d *Daemon) Start(ctx context.Context, started chan struct{}) error {
 			},
 		},
 		{
-			Name:           "notifier",
-			BackoffRestart: true,
-			Logger:         d.Logger,
-			Exclusive:      true,
-			DB:             d.DB,
-			LockID:         internal.Int64(notifications.LockID),
+			Name:      "notifier",
+			Logger:    d.Logger,
+			Exclusive: true,
+			DB:        d.DB,
+			LockID:    internal.Int64(notifications.LockID),
 			System: notifications.NewNotifier(notifications.NotifierOptions{
 				Logger:           d.Logger,
 				Subscriber:       d.Broker,
@@ -427,12 +422,11 @@ func (d *Daemon) Start(ctx context.Context, started chan struct{}) error {
 	}
 	if !d.DisableScheduler {
 		subsystems = append(subsystems, &Subsystem{
-			Name:           "scheduler",
-			BackoffRestart: true,
-			Logger:         d.Logger,
-			Exclusive:      true,
-			DB:             d.DB,
-			LockID:         internal.Int64(scheduler.LockID),
+			Name:      "scheduler",
+			Logger:    d.Logger,
+			Exclusive: true,
+			DB:        d.DB,
+			LockID:    internal.Int64(scheduler.LockID),
 			System: scheduler.NewScheduler(scheduler.Options{
 				Logger:           d.Logger,
 				WorkspaceService: d.WorkspaceService,

internal/daemon/subsystem.go

@@ -18,8 +18,6 @@ type (
 		Name string
 		// System is the underlying system to be invoked and supervised.
 		System Startable
-		// Backoff and restart subsystem in the event of an error
-		BackoffRestart bool
 		// Exclusive: permit only one instance of this subsystem on an OTF
 		// cluster
 		Exclusive bool
@@ -64,25 +62,22 @@ func (s *Subsystem) Start(ctx context.Context, g *errgroup.Group) error {
 		} else {
 			err = s.System.Start(ctx)
 		}
-		if err != nil {
-			return err
+		if ctx.Err() != nil {
+			// don't return an error if subsystem was terminated via a
+			// canceled context.
+			s.V(1).Info("gracefully shutdown subsystem", "name", s.Name)
+			return nil
 		}
-		s.V(1).Info("gracefully shutdown subsystem", "name", s.Name)
-		return nil
+		return err
 	}
-	if s.BackoffRestart {
-		// Backoff and retry whenever operation returns an error. If context is
-		// cancelled then it'll stop retrying and return the context error.
-		policy := backoff.WithContext(backoff.NewExponentialBackOff(), ctx)
-		g.Go(func() error {
-			return backoff.RetryNotify(op, policy, func(err error, next time.Duration) {
-				// re-open semaphore
-				s.Error(err, "restarting "+s.Name)
-			})
+	// Backoff and retry whenever operation returns an error. If context is
+	// cancelled then it'll stop retrying and return the context error.
+	policy := backoff.WithContext(backoff.NewExponentialBackOff(), ctx)
+	g.Go(func() error {
+		return backoff.RetryNotify(op, policy, func(err error, next time.Duration) {
+			s.Error(err, "restarting subsystem", "name", s.Name, "backoff", next)
 		})
-	} else {
-		g.Go(op)
-	}
+	})
 	s.V(1).Info("started subsystem", "name", s.Name)
 	return nil
 }

internal/daemon/subsystem_test.go

@@ -15,21 +15,18 @@ func TestSubsystem(t *testing.T) {
 
 	tests := []struct {
 		name      string
-		backoff   bool
 		exclusive bool
 	}{
-		{"default", false, false},
-		{"backoff", true, false},
-		{"backoff and wait and lock", true, true},
+		{"backoff", false},
+		{"backoff and wait and lock", true},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			sub := &Subsystem{
-				Name:           tt.name,
-				System:         &fakeStartable{},
-				Logger:         logr.Discard(),
-				BackoffRestart: tt.backoff,
-				Exclusive:      tt.exclusive,
+				Name:      tt.name,
+				System:    &fakeStartable{},
+				Logger:    logr.Discard(),
+				Exclusive: tt.exclusive,
 			}
 			if tt.exclusive {
 				sub.DB = &fakeWaitAndLock{}