Repository: triggermesh/aws-event-sources
Pull request #214 (2020-10-30 12:23:31)

Refactor Backoff to allows min below 1s
Also refactor tests to provide more control over the test logic


Patch:
diff --git a/pkg/adapter/awssqssource/adapter.go b/pkg/adapter/awssqssource/adapter.go
index 4795281..fe7b5f7 100644
--- a/pkg/adapter/awssqssource/adapter.go
+++ b/pkg/adapter/awssqssource/adapter.go
@@ -19,6 +19,7 @@ package awssqssource
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"go.uber.org/zap"
 
@@ -94,14 +95,13 @@ func (a *adapter) Start(ctx context.Context) error {
 	queueURL := *url.QueueUrl
 	a.logger.Infof("Listening to SQS queue at URL: %s", queueURL)
 
-	backoff := common.NewBackoff()
+	backoff := common.NewBackoff(1 * time.Millisecond)
 
 	err = backoff.Run(ctx.Done(), func(ctx context.Context) (bool, error) {
-		resetBackoff := false
 		messages, err := a.getMessages(queueURL)
 		if err != nil {
 			a.logger.Errorw("Failed to get messages from SQS queue", "error", err)
-			return resetBackoff, nil
+			return false, nil
 		}
 
 		for _, message := range messages {
@@ -117,10 +117,9 @@ func (a *adapter) Start(ctx context.Context) error {
 				a.logger.Errorw("Failed to delete message from SQS queue", "error", err)
 				continue
 			}
-			resetBackoff = true
 		}
 
-		return resetBackoff, nil
+		return true, nil
 	})
 
 	return err
diff --git a/pkg/adapter/common/backoff.go b/pkg/adapter/common/backoff.go
index 576be87..c9ec04f 100644
--- a/pkg/adapter/common/backoff.go
+++ b/pkg/adapter/common/backoff.go
@@ -19,6 +19,7 @@ package common
 import (
 	"context"
 	"math"
+	"sync/atomic"
 	"time"
 )
 
@@ -26,31 +27,32 @@ import (
 const (
 	expFactor = 2
 
-	minBackoff = 1 * time.Second
-	maxBackoff = 32 * time.Second
+	defaultMinBackoff = 1 * time.Second
+	defaultMaxBackoff = 32 * time.Second
 )
 
-// Backoff provides a simple exponential backoff mechanism
+// Backoff provides a simple exponential backoff mechanism.
 type Backoff struct {
-	step     int
+	step     *int32
 	factor   float64
 	min, max time.Duration
 }
 
-// RunFunc is a user function that polls data from a source and sends it as
-// a cloudevent to a sink.
-// RunFunc must return (bool, error) values where bool is true if poll backoff duration
-// must be reset and error is the result of function execution.
-type RunFunc func(context.Context) (bool, error)
+// RunFunc is a user function that polls data from a source and sends it as a
+// CloudEvent to a sink.
+// RunFunc must return (bool, error) values where bool is true if the poll
+// backoff duration must be reset, and error is the result of the function's
+// execution.
+type RunFunc func(context.Context) (bool /*reset*/, error /*exit*/)
 
-// NewBackoff accepts optional values for minimum and maximum wait period
-// and return new instance of Backoff structure
+// NewBackoff accepts optional values for minimum and maximum wait period and
+// returns a new instance of Backoff.
 func NewBackoff(args ...time.Duration) *Backoff {
 	backoff := &Backoff{
-		step:   0,
+		step:   new(int32),
 		factor: expFactor,
-		min:    minBackoff,
-		max:    maxBackoff,
+		min:    defaultMinBackoff,
+		max:    defaultMaxBackoff,
 	}
 
 	switch len(args) {
@@ -68,28 +70,36 @@ func NewBackoff(args ...time.Duration) *Backoff {
 	return backoff
 }
 
-// Duration can be used to get exponential backoff duration calculated for each new step
+// Duration returns the exponential backoff duration calculated for the current step.
 func (b *Backoff) Duration() time.Duration {
-	dur := time.Duration(float64(b.min)*math.Pow(b.factor, float64(b.step)) - float64(1*time.Second))
+	dur := time.Duration(float64(b.min)*math.Pow(b.factor, float64(atomic.LoadInt32(b.step))) - float64(b.min))
 
 	switch {
 	case dur < b.min:
-		b.step++
+		atomic.AddInt32(b.step, 1)
 		return b.min
 	case dur > b.max:
 		return b.max
 	default:
-		b.step++
+		atomic.AddInt32(b.step, 1)
 		return dur
 	}
 }
 
-// Run is a blocking function that executes RunFunc until stopCh receives the value
-// or function returns an error
+// Reset sets step counter to zero.
+func (b *Backoff) Reset() {
+	atomic.StoreInt32(b.step, 0)
+}
+
+// Run is a blocking function that executes RunFunc until stopCh receives a
+// value or fn returns an error.
 func (b *Backoff) Run(stopCh <-chan struct{}, fn RunFunc) error {
 	timer := time.NewTimer(0)
 	defer timer.Stop()
 
+	// FIXME(antoineco): never canceled until stopCh receives a value,
+	// after which the fn is never invoked again, so ctx does effectively
+	// nothing.
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -97,11 +107,13 @@ func (b *Backoff) Run(stopCh <-chan struct{}, fn RunFunc) error {
 		select {
 		case <-stopCh:
 			return nil
+
 		case <-timer.C:
 			reset, err := fn(ctx)
 			if err != nil {
 				return err
 			}
+
 			if reset {
 				b.Reset()
 			}
@@ -109,8 +121,3 @@ func (b *Backoff) Run(stopCh <-chan struct{}, fn RunFunc) error {
 		}
 	}
 }
-
-// Reset sets step counter to zero.
-func (b *Backoff) Reset() {
-	b.step = 0
-}


Test patch:
diff --git a/pkg/adapter/common/backoff_test.go b/pkg/adapter/common/backoff_test.go
index 2bfa18e..fedf368 100644
--- a/pkg/adapter/common/backoff_test.go
+++ b/pkg/adapter/common/backoff_test.go
@@ -18,7 +18,7 @@ package common
 
 import (
 	"context"
-	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -31,8 +31,8 @@ func TestNewBackoff(t *testing.T) {
 		gotMin, gotMax   time.Duration
 	}{
 		"defaults": {
-			gotMin: minBackoff,
-			gotMax: maxBackoff,
+			gotMin: defaultMinBackoff,
+			gotMax: defaultMaxBackoff,
 		},
 		"all correct": {
 			wantMin: time.Second,
@@ -42,14 +42,14 @@ func TestNewBackoff(t *testing.T) {
 		},
 		"min > default max": {
 			wantMin: time.Hour,
-			gotMin:  minBackoff,
-			gotMax:  maxBackoff,
+			gotMin:  defaultMinBackoff,
+			gotMax:  defaultMaxBackoff,
 		},
 		"min > max": {
 			wantMin: time.Hour,
-			gotMin:  minBackoff,
+			gotMin:  defaultMinBackoff,
 			wantMax: time.Minute,
-			gotMax:  maxBackoff,
+			gotMax:  defaultMaxBackoff,
 		},
 	}
 
@@ -70,25 +70,25 @@ func TestNewBackoff(t *testing.T) {
 
 func TestDuration(t *testing.T) {
 	testCases := map[string]struct {
-		step              int
+		step              int32
 		min, max, wantDur time.Duration
 	}{
 		"first step": {
 			step:    0,
 			min:     time.Second,
 			max:     time.Minute,
-			wantDur: time.Second,
+			wantDur: time.Second, // = min
 		},
 		"third step": {
 			step:    2,
 			min:     time.Second,
-			wantDur: 3 * time.Second,
+			wantDur: 3 * time.Second, // = 1s * factor(=2)^step(=2) - 1s
 		},
 		"tenth step": {
 			step:    9,
 			min:     time.Second,
 			max:     time.Minute,
-			wantDur: time.Minute,
+			wantDur: time.Minute, // = max
 		},
 	}
 
@@ -102,22 +102,23 @@ func TestDuration(t *testing.T) {
 				bo = NewBackoff(tc.min)
 			}
 
-			bo.step = tc.step
+			atomic.StoreInt32(bo.step, tc.step)
 			assert.Equal(t, tc.wantDur, bo.Duration())
 		})
 	}
 }
 
 func TestRun(t *testing.T) {
-	var counter int
-	start := time.Now()
-
 	testCases := map[string]struct {
 		fn         func(context.Context) (bool, error)
 		waitReturn bool
 		err        error
 	}{
-		"force stop with no errs": {
+		"force stop with no err": {
+			fn: func(ctx context.Context) (bool, error) {
+				// never terminates by itself
+				return true, nil
+			},
 			waitReturn: false,
 		},
 		"fn returns err": {
@@ -127,47 +128,83 @@ func TestRun(t *testing.T) {
 			waitReturn: true,
 			err:        assert.AnError,
 		},
-		"fn resets backoff duration": {
-			fn: func(ctx context.Context) (bool, error) {
-				counter++
-				if counter > 3 {
-					if time.Since(start).Round(time.Second) != 3*time.Second {
-						return true, fmt.Errorf("did we reset the backoff?")
-					}
-					// trigger Run() termination with expected error
-					return true, assert.AnError
-				}
-				// reset backoff duration to 1 sec for each iteration
-				return true, nil
-			},
-			waitReturn: true,
-			err:        assert.AnError,
-		},
 	}
 
-	errCh := make(chan error)
-	defer close(errCh)
-
 	for name, tc := range testCases {
-		stopCh := make(chan struct{})
 		t.Run(name, func(t *testing.T) {
-			bo := NewBackoff()
-			go func() {
-				err := bo.Run(stopCh, tc.fn)
-				errCh <- err
-			}()
-			if tc.waitReturn {
-				defer close(stopCh)
-			} else {
-				close(stopCh)
-			}
+			errCh := make(chan error)
+			defer close(errCh)
+
+			stopCh := make(chan struct{})
+
+			t.Run(name, func(t *testing.T) {
+				bo := NewBackoff()
+				go func() {
+					err := bo.Run(stopCh, tc.fn)
+					errCh <- err
+				}()
+
+				if tc.waitReturn {
+					defer close(stopCh)
+				} else {
+					close(stopCh)
+				}
 
-			err := <-errCh
-			if tc.err != nil {
-				assert.EqualError(t, err, tc.err.Error())
-			} else {
-				assert.NoError(t, err)
-			}
+				err := <-errCh
+				if tc.err != nil {
+					assert.EqualError(t, err, tc.err.Error())
+				} else {
+					assert.NoError(t, err)
+				}
+			})
 		})
 	}
+
+	t.Run("fn fails several times then resets backoff duration", func(t *testing.T) {
+		const minBackoff = 1 * time.Millisecond
+		const timesRunFnBeforeReset = 3
+
+		var returnVal bool
+
+		// latch is used to keep the function executions under control
+		// of the main test logic.
+		latch := make(chan struct{})
+
+		fn := func(ctx context.Context) (bool, error) {
+			latch <- struct{}{}
+			defer func() { latch <- struct{}{} }()
+
+			return returnVal, nil
+		}
+
+		bo := NewBackoff(minBackoff)
+
+		stopCh := make(chan struct{})
+		errCh := make(chan error)
+		defer close(errCh)
+
+		go func() {
+			errCh <- bo.Run(stopCh, fn)
+		}()
+
+		// cause timesRunFnBeforeReset fn executions
+		for i := 0; i < timesRunFnBeforeReset; i++ {
+			<-latch
+			<-latch
+		}
+		durationBeforeReset := bo.Duration()
+
+		returnVal = true // next run causes a reset
+		<-latch
+		<-latch
+
+		close(stopCh)
+		assert.NoError(t, <-errCh)
+		durationAfterReset := bo.Duration()
+
+		assert.Less(t, int64(durationAfterReset), int64(durationBeforeReset),
+			"Duration after reset is not less then before reset")
+
+		assert.Equal(t, durationAfterReset, minBackoff, "Backoff wasn't reset to its min value")
+	})
 }