Fix race condition (#426)

lovoo · Aug 3, 2023 · cd2141b · cd2141b
1 parent a14bfaf
commit cd2141b
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 1 deletion.
diff --git a/partition_table.go b/partition_table.go
@@ -3,6 +3,7 @@ package goka
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/Shopify/sarama"
@@ -137,27 +138,34 @@ func (p *PartitionTable) loadRestarting(ctx context.Context, stopAfterCatchup bo
 	var (
 		resetTimer *time.Timer
 		retries    int
+		retryMux   sync.Mutex
 	)
 
 	for {
 		err := p.load(ctx, stopAfterCatchup)
 		if err != nil {
 			p.log.Printf("Error while starting up: %v", err)
 
+			retryMux.Lock()
 			retries++
+			retryMux.Unlock()
 			if resetTimer != nil {
 				resetTimer.Stop()
 			}
 			resetTimer = time.AfterFunc(p.backoffResetTimeout, func() {
 				p.backoff.Reset()
+				retryMux.Lock()
 				retries = 0
+				retryMux.Unlock()
 			})
 		} else {
 			return nil
 		}
 
 		retryDuration := p.backoff.Duration()
+		retryMux.Lock()
 		p.log.Printf("Will retry in %.0f seconds (retried %d times so far)", retryDuration.Seconds(), retries)
+		retryMux.Unlock()
 		select {
 		case <-ctx.Done():
 			return nil

diff --git a/simple_backoff.go b/simple_backoff.go
@@ -1,6 +1,9 @@
 package goka
 
-import "time"
+import (
+	"sync"
+	"time"
+)
 
 // NewSimpleBackoff returns a simple backoff waiting the
 // specified duration longer each iteration until reset.
@@ -12,16 +15,22 @@ func NewSimpleBackoff(step time.Duration, max time.Duration) Backoff {
 }
 
 type simpleBackoff struct {
+	sync.Mutex
+
 	current time.Duration
 	step    time.Duration
 	max     time.Duration
 }
 
 func (b *simpleBackoff) Reset() {
+	b.Lock()
+	defer b.Unlock()
 	b.current = time.Duration(0)
 }
 
 func (b *simpleBackoff) Duration() time.Duration {
+	b.Lock()
+	defer b.Unlock()
 	value := b.current
 
 	if (b.current + b.step) <= b.max {