workflow/consumer: Simplify consumer pattern and change paused retry to consumer (#95)

* workflow/consumer: Clean up consumer pattern and convert paused retry to consumer

* make consume unexported

* add more unit tests and test cases

* clean up file naming

* clean up naming

Author: andrewwormald

adapters/kafkastreamer/kafka.go

@@ -22,8 +22,8 @@ type StreamConstructor struct {
 	brokers []string
 }
 
-func (s StreamConstructor) NewProducer(ctx context.Context, topic string) (workflow.Producer, error) {
-	return &Producer{
+func (s StreamConstructor) NewSender(ctx context.Context, topic string) (workflow.EventSender, error) {
+	return &Sender{
 		Topic: topic,
 		Writer: &kafka.Writer{
 			Addr:                   kafka.TCP(s.brokers...),
@@ -35,15 +35,15 @@ func (s StreamConstructor) NewProducer(ctx context.Context, topic string) (workf
 	}, nil
 }
 
-type Producer struct {
+type Sender struct {
 	Topic         string
 	Writer        *kafka.Writer
 	WriterTimeout time.Duration
 }
 
-var _ workflow.Producer = (*Producer)(nil)
+var _ workflow.EventSender = (*Sender)(nil)
 
-func (p *Producer) Send(ctx context.Context, foreignID string, statusType int, headers map[workflow.Header]string) error {
+func (p *Sender) Send(ctx context.Context, foreignID string, statusType int, headers map[workflow.Header]string) error {
 	for ctx.Err() == nil {
 		ctx, cancel := context.WithTimeout(ctx, p.WriterTimeout)
 		defer cancel()
@@ -76,12 +76,17 @@ func (p *Producer) Send(ctx context.Context, foreignID string, statusType int, h
 	return ctx.Err()
 }
 
-func (p *Producer) Close() error {
+func (p *Sender) Close() error {
 	return p.Writer.Close()
 }
 
-func (s StreamConstructor) NewConsumer(ctx context.Context, topic string, name string, opts ...workflow.ConsumerOption) (workflow.Consumer, error) {
-	var copts workflow.ConsumerOptions
+func (s StreamConstructor) NewReceiver(
+	ctx context.Context,
+	topic string,
+	name string,
+	opts ...workflow.ReceiverOption,
+) (workflow.EventReceiver, error) {
+	var copts workflow.ReceiverOptions
 	for _, opt := range opts {
 		opt(&copts)
 	}
@@ -101,25 +106,25 @@ func (s StreamConstructor) NewConsumer(ctx context.Context, topic string, name s
 		MaxWait:        time.Second,
 	})
 
-	return &Consumer{
+	return &Receiver{
 		topic:   topic,
 		name:    name,
 		reader:  kafkaReader,
 		options: copts,
 	}, nil
 }
 
-type Consumer struct {
+type Receiver struct {
 	topic   string
 	name    string
 	reader  *kafka.Reader
-	options workflow.ConsumerOptions
+	options workflow.ReceiverOptions
 }
 
-func (c *Consumer) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error) {
+func (r *Receiver) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error) {
 	var commit []kafka.Message
 	for ctx.Err() == nil {
-		m, err := c.reader.FetchMessage(ctx)
+		m, err := r.reader.FetchMessage(ctx)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -147,16 +152,16 @@ func (c *Consumer) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, err
 
 		return event,
 			func() error {
-				return c.reader.CommitMessages(ctx, commit...)
+				return r.reader.CommitMessages(ctx, commit...)
 			},
 			nil
 	}
 
 	return nil, nil, ctx.Err()
 }
 
-func (c *Consumer) Close() error {
-	return c.reader.Close()
+func (r *Receiver) Close() error {
+	return r.reader.Close()
 }
 
-var _ workflow.Consumer = (*Consumer)(nil)
+var _ workflow.EventReceiver = (*Receiver)(nil)

adapters/memstreamer/connector.go

@@ -53,7 +53,7 @@ type consumer struct {
 	cursorStore *cursorStore
 	cursorName  string
 	clock       clock.Clock
-	options     workflow.ConsumerOptions
+	options     workflow.ReceiverOptions
 }
 
 func (c *consumer) Recv(ctx context.Context) (*workflow.ConnectorEvent, workflow.Ack, error) {

adapters/memstreamer/memstreamer.go

@@ -51,7 +51,7 @@ type StreamConstructor struct {
 	cursorStore *cursorStore
 }
 
-func (s StreamConstructor) NewProducer(ctx context.Context, topic string) (workflow.Producer, error) {
+func (s StreamConstructor) NewSender(ctx context.Context, topic string) (workflow.EventSender, error) {
 	s.stream.mu.Lock()
 	defer s.stream.mu.Unlock()
 
@@ -63,11 +63,16 @@ func (s StreamConstructor) NewProducer(ctx context.Context, topic string) (workf
 	}, nil
 }
 
-func (s StreamConstructor) NewConsumer(ctx context.Context, topic string, name string, opts ...workflow.ConsumerOption) (workflow.Consumer, error) {
+func (s StreamConstructor) NewReceiver(
+	ctx context.Context,
+	topic string,
+	name string,
+	opts ...workflow.ReceiverOption,
+) (workflow.EventReceiver, error) {
 	s.stream.mu.Lock()
 	defer s.stream.mu.Unlock()
 
-	var options workflow.ConsumerOptions
+	var options workflow.ReceiverOptions
 	for _, opt := range opts {
 		opt(&options)
 	}
@@ -92,7 +97,7 @@ type Stream struct {
 	topic       string
 	name        string
 	clock       clock.Clock
-	options     workflow.ConsumerOptions
+	options     workflow.ReceiverOptions
 }
 
 func (s *Stream) Send(ctx context.Context, foreignID string, statusType int, headers map[workflow.Header]string) error {
@@ -147,8 +152,8 @@ func (s *Stream) Close() error {
 }
 
 var (
-	_ workflow.Producer = (*Stream)(nil)
-	_ workflow.Consumer = (*Stream)(nil)
+	_ workflow.EventSender   = (*Stream)(nil)
+	_ workflow.EventReceiver = (*Stream)(nil)
 )
 
 func newCursorStore() *cursorStore {

adapters/reflexstreamer/reflex.go

@@ -34,7 +34,7 @@ type constructor struct {
 	registerGapFiller sync.Once
 }
 
-func (c *constructor) NewProducer(ctx context.Context, topic string) (workflow.Producer, error) {
+func (c *constructor) NewSender(ctx context.Context, topic string) (workflow.EventSender, error) {
 	return &Producer{
 		topic:       topic,
 		writer:      c.writer,
@@ -79,8 +79,13 @@ func (p *Producer) Close() error {
 	return nil
 }
 
-func (c *constructor) NewConsumer(ctx context.Context, topic string, name string, opts ...workflow.ConsumerOption) (workflow.Consumer, error) {
-	var copts workflow.ConsumerOptions
+func (c *constructor) NewReceiver(
+	ctx context.Context,
+	topic string,
+	name string,
+	opts ...workflow.ReceiverOption,
+) (workflow.EventReceiver, error) {
+	var copts workflow.ReceiverOptions
 	for _, opt := range opts {
 		opt(&copts)
 	}
@@ -123,7 +128,7 @@ type Consumer struct {
 	cursor       reflex.CursorStore
 	reader       *sql.DB
 	streamClient reflex.StreamClient
-	options      workflow.ConsumerOptions
+	options      workflow.ReceiverOptions
 }
 
 func (c *Consumer) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error) {

adapters/reflexstreamer/reflex_test.go

@@ -31,7 +31,7 @@ func TestStreamerGapFiller(t *testing.T) {
 	jtest.RequireNil(t, err)
 
 	constructor := reflexstreamer.New(dbc, dbc, eventsTable, cTable.ToStore(dbc))
-	consumer, err := constructor.NewConsumer(context.Background(), "", "")
+	consumer, err := constructor.NewReceiver(context.Background(), "", "")
 	jtest.RequireNil(t, err)
 
 	for range 2 {

autopause.go

@@ -7,7 +7,6 @@ import (
 	"k8s.io/utils/clock"
 
 	"github.com/luno/workflow/internal/errorcounter"
-	"github.com/luno/workflow/internal/metrics"
 )
 
 // maybePause will either return a nil error if it has failed to pause the record and should be retried. A non-nil
@@ -20,7 +19,7 @@ func maybePause[Type any, Status StatusType](
 	originalErr error,
 	processName string,
 	run *Run[Type, Status],
-	logger *logger,
+	logger Logger,
 ) (paused bool, err error) {
 	// Only keep track of errors only if we need to
 	if pauseAfterErrCount == 0 {
@@ -37,7 +36,7 @@ func maybePause[Type any, Status StatusType](
 		return false, err
 	}
 
-	logger.maybeDebug(ctx, "paused record after exceeding allowed error count", map[string]string{
+	logger.Debug(ctx, "paused record after exceeding allowed error count", map[string]string{
 		"workflow_name": run.WorkflowName,
 		"foreign_id":    run.ForeignID,
 		"run_id":        run.RunID,
@@ -48,88 +47,94 @@ func maybePause[Type any, Status StatusType](
 	return true, nil
 }
 
-type autoPauseRetryConfig struct {
-	enabled bool
-	// limit determines the number of records in one lookup cycle.
-	limit int
-	// pollingFrequency is the frequency of the lookup cycle that looks up paused records that have met
-	// or exceeded the resumeAfter duration.
-	pollingFrequency time.Duration
-	// resumeAfter is the duration that the record should remain paused for.
-	resumeAfter time.Duration
-}
-
-func defaultAutoPauseRetryConfig() autoPauseRetryConfig {
-	return autoPauseRetryConfig{
-		enabled:          true,
-		limit:            10,
-		pollingFrequency: time.Minute,
-		resumeAfter:      time.Hour,
-	}
-}
+func pausedRecordsRetryConsumer[Type any, Status StatusType](w *Workflow[Type, Status]) {
+	role := makeRole(
+		w.Name(),
+		"paused",
+		"records",
+		"retry",
+		"consumer",
+	)
+
+	processName := makeRole(
+		"paused",
+		"records",
+		"retry",
+		"consumer",
+	)
+	w.run(role, processName, func(ctx context.Context) error {
+		topic := RunStateChangeTopic(w.Name())
+		stream, err := w.eventStreamer.NewReceiver(
+			ctx,
+			topic,
+			role,
+			WithReceiverPollFrequency(w.defaultOpts.pollingFrequency),
+		)
+		if err != nil {
+			return err
+		}
+		defer stream.Close()
 
-func autoRetryPausedRecordsForever[Type any, Status StatusType](w *Workflow[Type, Status]) {
-	role := makeRole(w.Name(), "paused-records-retry")
-	processName := role
+		lagAlert := w.pausedRecordsRetry.resumeAfter * 3
+		if lagAlert < time.Minute {
+			lagAlert = w.pausedRecordsRetry.resumeAfter + time.Minute*5
+		}
 
-	w.run(role, processName, func(ctx context.Context) error {
-		for {
-			err := retryPausedRecords(
-				ctx,
-				w.Name(),
-				w.recordStore.List,
+		return consume(
+			ctx,
+			w.Name(),
+			processName,
+			stream,
+			autoRetryConsumer(
+				w.recordStore.Lookup,
 				w.recordStore.Store,
 				w.clock,
-				processName,
-				w.autoPauseRetryConfig.limit,
-				w.autoPauseRetryConfig.resumeAfter,
-			)
-			if err != nil {
-				return err
-			}
-
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			case <-w.clock.After(w.autoPauseRetryConfig.pollingFrequency): // Slow and constant drip feed of paused records back into running state.
-				continue
-			}
-		}
+				w.pausedRecordsRetry.resumeAfter,
+			),
+			w.clock,
+			w.pausedRecordsRetry.resumeAfter,
+			lagAlert,
+		)
 	}, w.defaultOpts.errBackOff)
 }
 
-type listFunc func(ctx context.Context, workflowName string, offsetID int64, limit int, order OrderType, filters ...RecordFilter) ([]Record, error)
-
-func retryPausedRecords(
-	ctx context.Context,
-	workflowName string,
-	list listFunc,
+func autoRetryConsumer(
+	lookupFn lookupFunc,
 	store storeFunc,
 	clock clock.Clock,
-	processName string,
-	limit int,
 	retryInterval time.Duration,
-) error {
-	t0 := clock.Now()
-
-	rs, err := list(ctx, workflowName, 0, limit, OrderTypeAscending, FilterByRunState(RunStatePaused))
-	if err != nil {
-		return err
-	}
+) func(ctx context.Context, e *Event) error {
+	return func(ctx context.Context, e *Event) error {
+		record, err := lookupFn(ctx, e.ForeignID)
+		if err != nil {
+			return err
+		}
 
-	threshold := clock.Now().Add(-retryInterval)
-	for _, r := range rs {
-		if r.UpdatedAt.After(threshold) {
-			continue
+		threshold := clock.Now().Add(-retryInterval)
+		if record.UpdatedAt.After(threshold) {
+			return nil
 		}
 
-		controller := NewRunStateController(store, &r)
-		err := controller.Resume(ctx)
+		controller := NewRunStateController(store, record)
+		err = controller.Resume(ctx)
 		if err != nil {
 			return err
 		}
+
+		return nil
 	}
+}
 
-	metrics.ProcessLatency.WithLabelValues(workflowName, processName).Observe(clock.Since(t0).Seconds())
-	return nil
+type pausedRecordsRetry struct {
+	enabled bool
+	// resumeAfter is the duration that the record should remain paused for.
+	resumeAfter time.Duration
+}
+
+func defaultPausedRecordsRetry() pausedRecordsRetry {
+	resumeAfter := time.Hour
+	return pausedRecordsRetry{
+		enabled:     true,
+		resumeAfter: resumeAfter,
+	}
 }