Merge pull request #32125 from wojtek-t/fix_delta_fifo_race

Automatic merge from submit-queue

Fix race in DeltaFIFO

Fix #31981

@ncdc @deads2k @pmorie @gmarek

pkg/client/cache/delta_fifo.go

@@ -522,6 +522,18 @@ func (f *DeltaFIFO) syncKey(key string) error {
 		return nil
 	}
 
+	// If we are doing Resync() and there is already an event queued for that object,
+	// we ignore the Resync for it. This is to avoid the race, in which the resync
+	// comes with the previous value of object (since queueing an event for the object
+	// doesn't trigger changing the underlying store <knownObjects>.
+	id, err := f.KeyOf(obj)
+	if err != nil {
+		return KeyError{obj, err}
+	}
+	if len(f.items[id]) > 0 {
+		return nil
+	}
+
 	if err := f.queueActionLocked(Sync, obj); err != nil {
 		return fmt.Errorf("couldn't queue object: %v", err)
 	}

pkg/client/cache/delta_fifo_test.go

@@ -336,6 +336,29 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) {
 	}
 }
 
+func TestDeltaFIFO_UpdateResyncRace(t *testing.T) {
+	f := NewDeltaFIFO(
+		testFifoObjectKeyFunc,
+		nil,
+		keyLookupFunc(func() []testFifoObject {
+			return []testFifoObject{mkFifoObj("foo", 5)}
+		}),
+	)
+	f.Update(mkFifoObj("foo", 6))
+	f.Resync()
+
+	expectedList := []Deltas{
+		{{Updated, mkFifoObj("foo", 6)}},
+	}
+
+	for _, expected := range expectedList {
+		cur := Pop(f).(Deltas)
+		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
+			t.Errorf("Expected %#v, got %#v", e, a)
+		}
+	}
+}
+
 func TestDeltaFIFO_detectLineJumpers(t *testing.T) {
 	f := NewDeltaFIFO(testFifoObjectKeyFunc, nil, nil)