Add lock around progress setting.

pkg/sources/gcs/gcs.go

@@ -7,7 +7,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"sync/atomic"
 
 	diskbufferreader "github.com/bill-rich/disk-buffer-reader"
 	"github.com/go-errors/errors"
@@ -61,7 +60,8 @@ type Source struct {
 	processedObjects int32
 	cache            cache.Cache
 
-	sources.Progress
+	mu               sync.Mutex
+	sources.Progress // progress is not thread safe
 }
 
 // persistableCache is a wrapper around cache.Cache that allows
@@ -258,14 +258,16 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 }
 
 func (s *Source) setProgress(ctx context.Context, objName string) {
-	atomic.AddInt32(&s.processedObjects, 1)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	ctx.Logger().V(5).Info("setting progress for object", "object-name", objName)
+	s.processedObjects++
 
 	s.cache.Set(objName, objName)
-	processed := atomic.LoadInt32(&s.processedObjects)
-	s.Progress.SectionsCompleted = processed
+	s.Progress.SectionsCompleted = s.processedObjects
 	s.Progress.SectionsRemaining = int32(s.stats.numObjects)
-	s.Progress.PercentComplete = int64(float64(processed) / float64(s.stats.numObjects) * 100)
+	s.Progress.PercentComplete = int64(float64(s.processedObjects) / float64(s.stats.numObjects) * 100)
 }
 
 func (s *Source) completeProgress(ctx context.Context) {