fix: resubscribe subsystems when their subscription is terminated (#593)

Fixes #577 

The events broker can legitimately terminate the subscription of a
subscriber, e.g. when the subscriber's event queue is full up. Various
subsystems such as the scheduler would, in such a scenario, simply
shutdown. Which in the case of #577 would mean no more runs could be
started.

This PR changes the behaviour of the subsystems: when their subscription
is terminated, they instead return an error, which triggers the
subsystem to be restarted (using a backoff algorithm) rather than stop
completely.

Author: leg100

internal/logs/proxy.go

@@ -77,7 +77,7 @@ func (p *proxy) Start(ctx context.Context) error {
 			p.Error(err, "caching log chunk")
 		}
 	}
-	return nil
+	return pubsub.ErrSubscriptionTerminated
 }
 
 // GetChunk attempts to retrieve a chunk from the cache before falling back to

internal/notifications/notifier.go

@@ -72,7 +72,7 @@ func (s *Notifier) Start(ctx context.Context) error {
 			s.Error(err, "handling event", "event", event.Type)
 		}
 	}
-	return nil
+	return pubsub.ErrSubscriptionTerminated
 }
 
 func (s *Notifier) handle(ctx context.Context, event pubsub.Event) error {

internal/pubsub/broker.go

@@ -3,6 +3,7 @@ package pubsub
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -22,6 +23,10 @@ const (
 	subBufferSize = 100
 )
 
+// ErrSubscriptionTerminated is for use by subscribers to indicate that their
+// subscription has been terminated by the broker.
+var ErrSubscriptionTerminated = errors.New("broker terminated the subscription")
+
 type (
 
 	// Broker is a pubsub Broker implemented using postgres' listen/notify

internal/repo/purger.go

@@ -70,7 +70,7 @@ func (p *Purger) Start(ctx context.Context) error {
 			return err
 		}
 	}
-	return nil
+	return pubsub.ErrSubscriptionTerminated
 }
 
 func (p *Purger) deleteUnreferencedWebhooks(ctx context.Context) error {

internal/run/reporter.go

@@ -64,7 +64,7 @@ func (r *Reporter) Start(ctx context.Context) error {
 			return err
 		}
 	}
-	return nil
+	return pubsub.ErrSubscriptionTerminated
 }
 
 func (r *Reporter) handleRun(ctx context.Context, run *Run) error {

internal/scheduler/scheduler.go

@@ -151,5 +151,5 @@ func (s *scheduler) Start(ctx context.Context) error {
 			}
 		}
 	}
-	return nil
+	return pubsub.ErrSubscriptionTerminated
 }