Add a validAfter for retrying notifications

Author: Sergiu Ghitea

driver/sql/postgres/projector_aggregate_bench_test.go

@@ -182,6 +182,7 @@ func setup(
 		},
 		goengine.NopLogger,
 		driverSQL.NopMetrics,
+		0,
 	)
 	require.NoError(b, err, "failed to create aggregate projector")
 

driver/sql/projection.go

@@ -3,6 +3,7 @@ package sql
 import (
 	"context"
 	"database/sql"
+	"time"
 
 	"github.com/hellofresh/goengine"
 	"github.com/mailru/easyjson/jlexer"
@@ -12,8 +13,9 @@ import (
 type (
 	// ProjectionNotification is a representation of the data provided by database notify
 	ProjectionNotification struct {
-		No          int64  `json:"no"`
-		AggregateID string `json:"aggregate_id"`
+		No          int64     `json:"no"`
+		AggregateID string    `json:"aggregate_id"`
+		ValidAfter  time.Time `json:"valid_after"`
 	}
 
 	// ProjectionTrigger triggers the notification for processing

driver/sql/projector_aggregate.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"sync"
+	"time"
 
 	"github.com/hellofresh/goengine/aggregate"
 	"github.com/hellofresh/goengine/metadata"
@@ -24,6 +25,8 @@ type AggregateProjector struct {
 	db *sql.DB
 
 	logger goengine.Logger
+
+	retryDelay time.Duration
 }
 
 // NewAggregateProjector creates a new projector for a projection
@@ -36,6 +39,7 @@ func NewAggregateProjector(
 	projectionErrorHandler ProjectionErrorCallback,
 	logger goengine.Logger,
 	metrics Metrics,
+	retryDelay time.Duration,
 ) (*AggregateProjector, error) {
 	switch {
 	case db == nil:
@@ -76,6 +80,13 @@ func NewAggregateProjector(
 		return nil, err
 	}
 
+	if retryDelay == 0 {
+		retryDelay, err = time.ParseDuration("50ms")
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return &AggregateProjector{
 		backgroundProcessor:    processor,
 		executor:               executor,
@@ -85,6 +96,8 @@ func NewAggregateProjector(
 		db: db,
 
 		logger: logger,
+
+		retryDelay: retryDelay,
 	}, nil
 }
 
@@ -153,6 +166,7 @@ func (a *AggregateProjector) processNotification(
 		return nil
 	case errorRetry:
 		a.logger.Debug("ProcessHandler->ErrorHandler: re-queueing notification", logFields)
+		notification.ValidAfter = time.Now().Add(a.retryDelay)
 		return queue(ctx, notification)
 	}
 

strategy/json/sql/postgres/manager.go

@@ -2,6 +2,7 @@ package postgres
 
 import (
 	"database/sql"
+	"time"
 
 	"github.com/hellofresh/goengine"
 	driverSQL "github.com/hellofresh/goengine/driver/sql"
@@ -120,6 +121,7 @@ func (m *SingleStreamManager) NewAggregateProjector(
 	projection goengine.Projection,
 	projectionErrorHandler driverSQL.ProjectionErrorCallback,
 	useLockedField bool,
+	retryDelay time.Duration,
 ) (*driverSQL.AggregateProjector, error) {
 	eventStore, err := m.NewEventStore()
 	if err != nil {
@@ -151,5 +153,6 @@ func (m *SingleStreamManager) NewAggregateProjector(
 		projectionErrorHandler,
 		m.logger,
 		m.metrics,
+		retryDelay,
 	)
 }

test/projector_aggregate_integration_test.go

@@ -111,6 +111,7 @@ func (s *aggregateProjectorTestSuite) TestRunAndListen() {
 		},
 		s.GetLogger(),
 		s.Metrics,
+		0,
 	)
 	s.Require().NoError(err, "failed to create projector")
 
@@ -209,6 +210,7 @@ func (s *aggregateProjectorTestSuite) TestRunAndListen() {
 			},
 			s.GetLogger(),
 			s.Metrics,
+			0,
 		)
 		s.Require().NoError(err, "failed to create projector")
 
@@ -264,6 +266,7 @@ func (s *aggregateProjectorTestSuite) TestRun() {
 		},
 		s.GetLogger(),
 		s.Metrics,
+		0,
 	)
 	s.Require().NoError(err, "failed to create projector")