Fix exponent overflow

pkg/controller/nodes/task/backoff/controller.go

@@ -37,35 +37,18 @@ func (m *Controller) GetOrCreateHandler(ctx context.Context, key string, backOff
 	} else {
 		logger.Infof(ctx, "The back-off handler for [%v] has been created.\n", key)
 	}
+
 	if ret, casted := h.(*ComputeResourceAwareBackOffHandler); casted {
 		return ret
 	}
+
 	return nil
 }
 
 func (m *Controller) GetBackOffHandler(key string) (*ComputeResourceAwareBackOffHandler, bool) {
 	return m.backOffHandlerMap.Get(key)
 }
 
-func (m *Controller) CreateBackOffHandler(ctx context.Context, key string, backOffBaseSecond int, maxBackOffDuration time.Duration) *ComputeResourceAwareBackOffHandler {
-	m.backOffHandlerMap.Set(key, &ComputeResourceAwareBackOffHandler{
-		SimpleBackOffBlocker: &SimpleBackOffBlocker{
-			Clock:              m.Clock,
-			BackOffBaseSecond:  backOffBaseSecond,
-			BackOffExponent:    0,
-			NextEligibleTime:   m.Clock.Now(),
-			MaxBackOffDuration: maxBackOffDuration,
-		},
-		ComputeResourceCeilings: &ComputeResourceCeilings{
-			computeResourceCeilings: v1.ResourceList{},
-		},
-	})
-	h, _ := m.backOffHandlerMap.Get(key)
-	h.reset()
-	logger.Infof(ctx, "The back-off handler for [%v] has been created.\n", key)
-	return h
-}
-
 func ComposeResourceKey(o k8s.Resource) string {
 	return fmt.Sprintf("%v,%v", o.GroupVersionKind().String(), o.GetNamespace())
 }

pkg/controller/nodes/task/backoff/handler.go

@@ -5,6 +5,7 @@ import (
 	"math"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/lyft/flyteplugins/go/tasks/errors"
@@ -24,33 +25,49 @@ var (
 type SimpleBackOffBlocker struct {
 	Clock              clock.Clock
 	BackOffBaseSecond  int
-	BackOffExponent    int
-	NextEligibleTime   time.Time
 	MaxBackOffDuration time.Duration
+
+	// Mutable fields
+	syncObj          sync.RWMutex
+	BackOffExponent  int
+	NextEligibleTime time.Time
 }
 
 func (b *SimpleBackOffBlocker) isBlocking(t time.Time) bool {
+	b.syncObj.RLock()
+	defer b.syncObj.RUnlock()
+
 	return !b.NextEligibleTime.Before(t)
 }
 
 func (b *SimpleBackOffBlocker) getBlockExpirationTime() time.Time {
+	b.syncObj.RLock()
+	defer b.syncObj.RUnlock()
+
 	return b.NextEligibleTime
 }
 
 func (b *SimpleBackOffBlocker) reset() {
+	b.syncObj.Lock()
+	defer b.syncObj.Unlock()
+
 	b.BackOffExponent = 0
 	b.NextEligibleTime = b.Clock.Now()
 }
 
 func (b *SimpleBackOffBlocker) backOff() time.Duration {
+	b.syncObj.Lock()
+	defer b.syncObj.Unlock()
+
 	backOffDuration := time.Duration(time.Second.Nanoseconds() * int64(math.Pow(float64(b.BackOffBaseSecond), float64(b.BackOffExponent))))
 
 	if backOffDuration > b.MaxBackOffDuration {
 		backOffDuration = b.MaxBackOffDuration
+	} else {
+		b.BackOffExponent++
 	}
 
 	b.NextEligibleTime = b.Clock.Now().Add(backOffDuration)
-	b.BackOffExponent++
 	return backOffDuration
 }
 

pkg/controller/nodes/task/backoff/handler_test.go

@@ -7,6 +7,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/magiconair/properties/assert"
+
 	taskErrors "github.com/lyft/flyteplugins/go/tasks/errors"
 	stdlibErrors "github.com/lyft/flytestdlib/errors"
 	v1 "k8s.io/api/core/v1"
@@ -450,7 +452,7 @@ func TestSimpleBackOffBlocker_backOff(t *testing.T) {
 		},
 		{name: "backoff should saturate",
 			fields:       fields{Clock: tc, BackOffBaseSecond: 2, BackOffExponent: 10, NextEligibleTime: tc.Now(), MaxBackOffDuration: maxBackOffDuration},
-			wantExponent: 11,
+			wantExponent: 10,
 			wantDuration: maxBackOffDuration,
 		},
 	}
@@ -472,4 +474,20 @@ func TestSimpleBackOffBlocker_backOff(t *testing.T) {
 			}
 		})
 	}
+
+	t.Run("backoff many times after maxBackOffDuration is hit", func(t *testing.T) {
+		b := &SimpleBackOffBlocker{
+			Clock:              tc,
+			BackOffBaseSecond:  2,
+			BackOffExponent:    10,
+			NextEligibleTime:   tc.Now(),
+			MaxBackOffDuration: maxBackOffDuration,
+		}
+
+		for i := 0; i < 10; i++ {
+			backOffDuration := b.backOff()
+			assert.Equal(t, maxBackOffDuration, backOffDuration)
+			assert.Equal(t, 10, b.BackOffExponent)
+		}
+	})
 }