Fix a data race in watcher.go

provider/go.mod

@@ -192,7 +192,7 @@ require (
 	github.com/zclconf/go-cty v1.13.2 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.starlark.net v0.0.0-20230525235612-a134d8f9ddca // indirect
-	go.uber.org/atomic v1.11.0 // indirect
+	go.uber.org/atomic v1.11.0
 	gocloud.dev v0.36.0 // indirect
 	gocloud.dev/secrets/hashivault v0.27.0 // indirect
 	golang.org/x/net v0.21.0

provider/pkg/watcher/watcher.go

@@ -114,9 +114,8 @@ func (ow *ObjectWatcher) watch(
 	var obj *unstructured.Unstructured
 	results := make(chan result)
 	poll := func() {
-		var err error
-		obj, err = ow.pollFunc()
-		results <- result{Obj: obj, Err: err}
+		o, err := ow.pollFunc()
+		results <- result{Obj: o, Err: err}
 	}
 
 	wait := 500 * time.Millisecond
@@ -129,6 +128,7 @@ func (ow *ObjectWatcher) watch(
 		case <-ow.ctx.Done():
 			return cancellationErr(ow.objName, obj)
 		case res := <-results:
+			obj = res.Obj
 			if stop, err := until(res.Obj, res.Err); err != nil {
 				return err
 			} else if stop {
@@ -160,13 +160,17 @@ type result struct {
 // --------------------------------------------------------------------------
 
 func timeoutErr(name string, obj *unstructured.Unstructured) error {
-	return &watchError{object: obj,
-		message: fmt.Sprintf("Timeout occurred polling for '%s'", name)}
+	return &watchError{
+		object:  obj,
+		message: fmt.Sprintf("Timeout occurred polling for '%s'", name),
+	}
 }
 
 func cancellationErr(name string, obj *unstructured.Unstructured) error {
-	return &watchError{object: obj,
-		message: fmt.Sprintf("Resource operation was cancelled for '%s'", name)}
+	return &watchError{
+		object:  obj,
+		message: fmt.Sprintf("Resource operation was cancelled for '%s'", name),
+	}
 }
 
 type watchError struct {

provider/pkg/watcher/watcher_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 	"time"
 
+	"go.uber.org/atomic"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 
@@ -72,27 +73,27 @@ func Test_WatchUntil_PollFuncTimeout(t *testing.T) {
 	testCompleted := make(chan struct{})
 	for _, test := range timeoutTests {
 		go func(test timeoutTest) {
-			pollFuncCalls, watchFuncCalls := 0, 0
+			pollFuncCalls, watchFuncCalls := atomic.NewInt32(0), atomic.NewInt32(0)
 			err := testObjWatcher(
 				context.Background(),
 				func() (*unstructured.Unstructured, error) {
-					pollFuncCalls++
+					pollFuncCalls.Inc()
 					return test.pollFunc()
 				}).
 				WatchUntil(
 					func(obj *unstructured.Unstructured) bool {
-						watchFuncCalls++
+						watchFuncCalls.Inc()
 						return test.predicate(obj)
 					},
 					test.timeout)
 			if err == nil || !strings.HasPrefix(err.Error(), timeoutErrPrefix) {
 				t.Errorf("%s: Polling should have timed out", test.name)
 			}
-			if !test.targetPollFuncCalls(pollFuncCalls) {
-				t.Errorf("%s: Got %d poll function calls, which did not satisfy the test predicate", test.name, pollFuncCalls)
+			if !test.targetPollFuncCalls(int(pollFuncCalls.Load())) {
+				t.Errorf("%s: Got %d poll function calls, which did not satisfy the test predicate", test.name, pollFuncCalls.Load())
 			}
-			if !test.targetWatchFuncCalls(watchFuncCalls) {
-				t.Errorf("%s: Got %d watch function calls, which did not satisfy the test predicate", test.name, watchFuncCalls)
+			if !test.targetWatchFuncCalls(int(watchFuncCalls.Load())) {
+				t.Errorf("%s: Got %d watch function calls, which did not satisfy the test predicate", test.name, watchFuncCalls.Load())
 			}
 			testCompleted <- struct{}{}
 		}(test)
@@ -119,7 +120,6 @@ func Test_WatchUntil_Success(t *testing.T) {
 				return true // Always true.
 			},
 			1*time.Second)
-
 	if err != nil {
 		t.Error("Expected watch to terminate without error")
 	}
@@ -137,7 +137,6 @@ func Test_RetryUntil_Success(t *testing.T) {
 				return nil // Always succeeds.
 			},
 			1*time.Second)
-
 	if err != nil {
 		t.Error("Expected watch to terminate without error")
 	}