feat: introduce exp. backoff for polling (#52)

- Introduce active poll percent configuration to determine the
  percentage of workers that should actively poll as per the specified
  poll interval irrespective of whether a job was found or not.
- For a non-active poll worker, when no job is found, use exponential
  backoff to determine the poll delay for next attempt.
- Fix bug in exponential backoff calculation where it would return 0 as
  the backoff duration for high attempt number such as 70.
- Add instance name to otelsql instrumentation and use nrpgx driver for
  default PostgreSQL client (was missed).

compass.yaml.example

@@ -59,6 +59,7 @@ worker:
     enabled: true
     worker_count: 1
     poll_interval: 1s
+    active_poll_percent: 20
     pgq:
         host: localhost
         port: 5432

internal/store/postgres/postgres.go

@@ -20,7 +20,6 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgerrcode"
-	_ "github.com/jackc/pgx/v4/stdlib"
 	"github.com/jmoiron/sqlx"
 	_ "github.com/newrelic/go-agent/v3/integrations/nrpgx" // register instrumented DB driver
 	"go.nhat.io/otelsql"
@@ -96,11 +95,12 @@ func (c *Client) Close() error {
 // NewClient initializes database connection
 func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 	driverName, err := otelsql.Register(
-		"pgx",
+		"nrpgx",
 		otelsql.TraceQueryWithoutArgs(),
 		otelsql.TraceRowsClose(),
 		otelsql.TraceRowsAffected(),
 		otelsql.WithSystem(semconv.DBSystemPostgreSQL),
+		otelsql.WithInstanceName("default"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("register otelsql: %w", err)

internal/workermanager/worker_manager.go

@@ -35,11 +35,12 @@ type Worker interface {
 }
 
 type Config struct {
-	Enabled        bool          `mapstructure:"enabled"`
-	WorkerCount    int           `mapstructure:"worker_count" default:"3"`
-	PollInterval   time.Duration `mapstructure:"poll_interval" default:"500ms"`
-	PGQ            pgq.Config    `mapstructure:"pgq"`
-	JobManagerPort int           `mapstructure:"job_manager_port"`
+	Enabled           bool          `mapstructure:"enabled"`
+	WorkerCount       int           `mapstructure:"worker_count" default:"3"`
+	PollInterval      time.Duration `mapstructure:"poll_interval" default:"500ms"`
+	ActivePollPercent float64       `mapstructure:"active_poll_percent" default:"20"`
+	PGQ               pgq.Config    `mapstructure:"pgq"`
+	JobManagerPort    int           `mapstructure:"job_manager_port"`
 }
 
 type Deps struct {
@@ -58,6 +59,7 @@ func New(ctx context.Context, deps Deps) (*Manager, error) {
 	w, err := worker.New(
 		workermw.WithJobProcessorInstrumentation()(processor),
 		worker.WithRunConfig(cfg.WorkerCount, cfg.PollInterval),
+		worker.WithActivePollPercent(cfg.ActivePollPercent),
 		worker.WithLogger(deps.Logger),
 	)
 	if err != nil {

pkg/worker/backoff_strategy.go

@@ -69,7 +69,12 @@ type ExponentialBackoff struct {
 }
 
 func (b *ExponentialBackoff) Backoff(attempt int) time.Duration {
-	duration := b.InitialDelay * time.Duration(math.Pow(b.Multiplier, float64(attempt-1)))
+	limit := (float64)(math.MaxInt64) / (float64)(b.InitialDelay)
+	multiplier := math.Pow(b.Multiplier, (float64)(attempt-1))
+	if multiplier > limit {
+		multiplier = limit
+	}
+	duration := b.InitialDelay * time.Duration(multiplier)
 
 	if b.MaxDelay > 0 && duration > b.MaxDelay {
 		duration = b.MaxDelay

pkg/worker/backoff_strategy_test.go

@@ -199,6 +199,16 @@ func TestExponentialBackoff(t *testing.T) {
 			attempt:  11,
 			expected: time.Second * 15,
 		},
+		7: {
+			b: &ExponentialBackoff{
+				Multiplier:   4,
+				InitialDelay: time.Second * 1,
+				MaxDelay:     time.Second * 10,
+				Jitter:       1,
+			},
+			attempt:  111,
+			expected: time.Second * 15,
+		},
 	}
 	for i, tc := range cases {
 		assert.Equal(t, tc.expected, tc.b.Backoff(tc.attempt), "test[%d]", i)

pkg/worker/pgq/pgq_processor.go

@@ -36,6 +36,7 @@ func NewProcessor(ctx context.Context, cfg Config) (*Processor, error) {
 		otelsql.TraceRowsClose(),
 		otelsql.TraceRowsAffected(),
 		otelsql.WithSystem(semconv.DBSystemPostgreSQL),
+		otelsql.WithInstanceName("pgq"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("new pgq processor: %w", err)
@@ -98,7 +99,7 @@ func (p *Processor) Process(ctx context.Context, types []string, fn worker.JobEx
 		job, err := p.pickupJob(ctx, tx, types)
 		if err != nil {
 			if errors.Is(err, sql.ErrNoRows) {
-				return nil
+				return fmt.Errorf("pickup job: %w", worker.ErrNoJob)
 			}
 			return fmt.Errorf("pickup job: %w", err)
 		}

pkg/worker/pgq/pgq_processor_test.go

@@ -154,7 +154,7 @@ func (s *ProcessorTestSuite) TestProcess() {
 			s.Fail("unexpected job invocation")
 			return job
 		})
-		s.NoError(err)
+		s.ErrorIs(err, worker.ErrNoJob)
 	})
 
 	s.Run("ProcessOnlyGivenTypes", func() {
@@ -168,7 +168,7 @@ func (s *ProcessorTestSuite) TestProcess() {
 			s.Fail("unexpected job invocation")
 			return job
 		})
-		s.NoError(err)
+		s.ErrorIs(err, worker.ErrNoJob)
 	})
 
 	s.Run("ProcessOnlyReadyJobs", func() {
@@ -184,7 +184,7 @@ func (s *ProcessorTestSuite) TestProcess() {
 			s.Fail("unexpected job invocation")
 			return job
 		})
-		s.NoError(err)
+		s.ErrorIs(err, worker.ErrNoJob)
 	})
 
 	s.Run("JobProcessedSuccessfully", func() {
@@ -296,7 +296,7 @@ func (s *ProcessorTestSuite) TestProcess() {
 				s.Fail("unexpected job invocation")
 				return job
 			})
-			s.NoError(err)
+			s.ErrorIs(err, worker.ErrNoJob)
 			return job
 		})
 		s.NoError(err)

pkg/worker/worker.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"sync"
 	"time"
 
@@ -14,12 +15,14 @@ var (
 	ErrTypeExists  = errors.New("handler for given job type exists")
 	ErrUnknownType = errors.New("job type is invalid")
 	ErrJobExists   = errors.New("job with id exists")
+	ErrNoJob       = errors.New("no job found")
 )
 
 // Worker provides asynchronous job processing using a job processor.
 type Worker struct {
-	workers      int
-	pollInterval time.Duration
+	workers           int
+	pollInterval      time.Duration
+	activePollPercent float64
 
 	processor JobProcessor
 	logger    log.Logger
@@ -87,13 +90,15 @@ func (w *Worker) Run(baseCtx context.Context) error {
 	ctx, cancel := context.WithCancel(baseCtx)
 	defer cancel()
 
+	activePollWorkers := (int)(math.Ceil((float64)(w.workers) * w.activePollPercent / 100))
+
 	var wg sync.WaitGroup
 	wg.Add(w.workers)
 	for i := 0; i < w.workers; i++ {
 		go func(id int) {
 			defer wg.Done()
 
-			w.runWorker(ctx)
+			w.runWorker(ctx, id < activePollWorkers)
 			w.logger.Info("worker exited", "worker_id", id)
 		}(i)
 	}
@@ -103,26 +108,46 @@ func (w *Worker) Run(baseCtx context.Context) error {
 	return cleanupCtxErr(ctx.Err())
 }
 
-func (w *Worker) runWorker(ctx context.Context) {
-	ticker := time.NewTicker(w.pollInterval)
-	defer ticker.Stop()
+func (w *Worker) runWorker(ctx context.Context, activePoll bool) {
+	timer := time.NewTimer(w.pollInterval)
+	defer timer.Stop()
+
+	var backoff BackoffStrategy = ConstBackoff{Delay: w.pollInterval}
+	if !activePoll {
+		backoff = &ExponentialBackoff{
+			Multiplier:   1.6,
+			InitialDelay: w.pollInterval,
+			MaxDelay:     5 * time.Second,
+			Jitter:       0.5,
+		}
+	}
 
+	pollAttempt := 1
 	for {
 		select {
 		case <-ctx.Done():
 			return
 
-		case <-ticker.C:
+		case <-timer.C:
 			types := w.getTypes()
 			if len(types) == 0 {
 				w.logger.Warn("no job-handler registered, skipping processing")
 				continue
 			}
 
-			w.logger.Debug("looking for a job", "types", types)
-			if err := w.processor.Process(ctx, types, w.processJob); err != nil {
-				w.logger.Error("process job failed", "error", err)
+			w.logger.Debug("looking for a job", "types", types, "active_poll", activePoll)
+			switch err := w.processor.Process(ctx, types, w.processJob); {
+			case err != nil && errors.Is(err, ErrNoJob):
+				pollAttempt++
+
+			case err != nil:
+				w.logger.Error("process job failed", "err", err)
+				pollAttempt = 1
+
+			default:
+				pollAttempt = 1
 			}
+			timer.Reset(backoff.Backoff(pollAttempt))
 		}
 	}
 }

pkg/worker/worker_option.go

@@ -39,9 +39,17 @@ func WithRunConfig(workers int, pollInterval time.Duration) Option {
 	}
 }
 
+func WithActivePollPercent(pct float64) Option {
+	return func(w *Worker) error {
+		w.activePollPercent = pct
+		return nil
+	}
+}
+
 func withDefaults(opts []Option) []Option {
 	return append([]Option{
 		WithLogger(nil),
 		WithRunConfig(1, 1*time.Second),
+		WithActivePollPercent(20),
 	}, opts...)
 }