Fix race condition causing negative WaitGroup counter in delta watches

Author: krhitesh7

pkg/cache/v3/simple.go

@@ -563,8 +563,9 @@ func (cache *snapshotCache) respondSOTWWatches(ctx context.Context, info *status
 	// sending them in the correct order. Go's default implementation
 	// of maps are randomized order when ranged over.
 	if cache.ads {
-		info.orderResponseWatches()
-		for _, key := range info.orderedWatches {
+		// Create a local copy of ordered watches to avoid race conditions
+		orderedWatches := info.getOrderedWatches()
+		for _, key := range orderedWatches {
 			err := respond(info.watches[key.ID], key.ID)
 			if err != nil {
 				return err
@@ -600,11 +601,12 @@ func (cache *snapshotCache) respondDeltaWatches(ctx context.Context, info *statu
 	// of maps are randomized order when ranged over.
 	if cache.ads {
 		start := time.Now()
-		info.orderResponseDeltaWatches()
+		// Create a local copy of ordered watches to avoid race conditions
+		orderedWatches := info.getOrderedDeltaWatches()
 		// Use a buffered channel to safely collect ids to delete from multiple goroutines.
-		toDeleteCh := make(chan int64, len(info.orderedDeltaWatches))
+		toDeleteCh := make(chan int64, len(orderedWatches))
 		wg := sync.WaitGroup{}
-		for _, k := range info.orderedDeltaWatches {
+		for _, k := range orderedWatches {
 			wg.Add(1)
 			watch := info.deltaWatches[k.ID]
 			// One goroutine for each client request awaiting response

pkg/cache/v3/status.go

@@ -196,6 +196,26 @@ func (info *statusInfo) orderResponseWatches() {
 	sort.Sort(info.orderedWatches)
 }
 
+// getOrderedWatches creates and returns a local copy of ordered watch keys.
+// This avoids race conditions by not using a shared struct field.
+func (info *statusInfo) getOrderedWatches() keys {
+	orderedWatches := make(keys, len(info.watches))
+
+	var index int
+	for id, watch := range info.watches {
+		orderedWatches[index] = key{
+			ID:      id,
+			TypeURL: watch.Request.GetTypeUrl(),
+		}
+		index++
+	}
+
+	// Sort our list which we can use in the SetSnapshot functions.
+	// This is only run when we enable ADS on the cache.
+	sort.Sort(orderedWatches)
+	return orderedWatches
+}
+
 // orderResponseDeltaWatches will track a list of delta watch keys and order them if
 // true is passed.
 func (info *statusInfo) orderResponseDeltaWatches() {
@@ -214,3 +234,23 @@ func (info *statusInfo) orderResponseDeltaWatches() {
 	// This is only run when we enable ADS on the cache.
 	sort.Sort(info.orderedDeltaWatches)
 }
+
+// getOrderedDeltaWatches creates and returns a local copy of ordered delta watch keys.
+// This avoids race conditions by not using a shared struct field.
+func (info *statusInfo) getOrderedDeltaWatches() keys {
+	orderedWatches := make(keys, len(info.deltaWatches))
+
+	var index int
+	for id, deltaWatch := range info.deltaWatches {
+		orderedWatches[index] = key{
+			ID:      id,
+			TypeURL: deltaWatch.Request.GetTypeUrl(),
+		}
+		index++
+	}
+
+	// Sort our list which we can use in the SetSnapshot functions.
+	// This is only run when we enable ADS on the cache.
+	sort.Sort(orderedWatches)
+	return orderedWatches
+}