From 796e1455bc5d77b9059ea54ca243b52fe236c549 Mon Sep 17 00:00:00 2001 From: lanfix Date: Tue, 8 Sep 2026 15:42:52 +0300 Subject: [PATCH 1/3] :bug: priorityqueue: Fix the data race with ReadAt field; --- pkg/controller/priorityqueue/priorityqueue.go | 40 +++++++++---- .../priorityqueue/priorityqueue_test.go | 60 +++++++++++++++++++ 2 files changed, 88 insertions(+), 12 deletions(-) diff --git a/pkg/controller/priorityqueue/priorityqueue.go b/pkg/controller/priorityqueue/priorityqueue.go index dd267cd45a..1d81bd0aca 100644 --- a/pkg/controller/priorityqueue/priorityqueue.go +++ b/pkg/controller/priorityqueue/priorityqueue.go @@ -537,20 +537,36 @@ func (w *priorityqueue[T]) logState() { if !w.log.V(5).Enabled() { continue } - w.lock.Lock() - items := make([]*item[T], 0, len(w.items)) - w.waiting.Ascend(func(item *item[T]) bool { - items = append(items, item) - return true - }) - w.ready.Ascend(func(item *item[T]) bool { - items = append(items, item) - return true - }) - w.lock.Unlock() - w.log.V(5).Info("workqueue_items", "items", items) + w.log.V(5).Info("workqueue_items", "items", w.cloneItems()) + } +} + +// cloneItems returns a by-value deep copy of every queued item that is safe to +// use after w.lock is released. logState serializes the result outside the lock +// while other goroutines keep mutating the live items (handleWaitingItems sets +// ReadyAt to nil as an item becomes ready, lockedAddWithOpts reassigns it on the +// update path), so handing the *item[T] pointers straight to the logger is a +// data race. The race can crash the process from within encoding/json when a +// ReadyAt pointer is observed non-nil and then dereferenced after it went nil: +// "value method time.Time.MarshalJSON called using nil *Time pointer". +func (w *priorityqueue[T]) cloneItems() []item[T] { + w.lock.Lock() + defer w.lock.Unlock() + + items := make([]item[T], 0, len(w.items)) + appendItem := func(it *item[T]) bool { + clone := *it + if it.ReadyAt != nil { + clone.ReadyAt = new(*it.ReadyAt) + } + items = append(items, clone) + return true } + w.waiting.Ascend(appendItem) + w.ready.Ascend(appendItem) + + return items } func lessWaiting[T comparable](a, b *item[T]) bool { diff --git a/pkg/controller/priorityqueue/priorityqueue_test.go b/pkg/controller/priorityqueue/priorityqueue_test.go index bb6b658c86..f6547443cc 100644 --- a/pkg/controller/priorityqueue/priorityqueue_test.go +++ b/pkg/controller/priorityqueue/priorityqueue_test.go @@ -17,6 +17,7 @@ limitations under the License. package priorityqueue import ( + "encoding/json" "fmt" "math/rand/v2" "strconv" @@ -1068,3 +1069,62 @@ func TestPriorityQueueDoesNotDeadlockOnShutdownWhileHandingOutItems(t *testing.T } } } + +// TestPriorityQueueLogStateDoesNotRaceOnReadyAt is a regression test for a data +// race in logState. It used to collect the []*item[T] pointers under w.lock and +// then hand them to the logger after releasing the lock, while handleWaitingItems +// (toMove.ReadyAt = nil) and lockedAddWithOpts (item.ReadyAt = readyAt) kept +// mutating those same items. Serializing a ReadyAt pointer that races from +// non-nil to nil crashes the process inside encoding/json with +// "value method time.Time.MarshalJSON called using nil *Time pointer". +// cloneItems now returns an independent by-value copy taken under the lock; this +// test reproduces the concurrent access and must be run with -race. +func TestPriorityQueueLogStateDoesNotRaceOnReadyAt(t *testing.T) { + t.Parallel() + + q, _ := newQueue() + defer q.ShutDown() + + const keySpace = 20 + stop := make(chan struct{}) + var wg sync.WaitGroup + + // Writers: churn items through waiting -> ready (handleWaitingItems clears + // ReadyAt) and through the update path (lockedAddWithOpts reassigns ReadyAt). + for range 8 { + wg.Go(func() { + for { + select { + case <-stop: + return + default: + } + q.AddWithOpts(AddOpts{ + After: time.Duration(rand.IntN(3)) * time.Millisecond, + Priority: new(rand.IntN(5)), + }, strconv.Itoa(rand.IntN(keySpace))) + } + }) + } + + // Readers: do what logState does - snapshot the queue and serialize it. + for range 4 { + wg.Go(func() { + for { + select { + case <-stop: + return + default: + } + if _, err := json.Marshal(q.cloneItems()); err != nil { + t.Errorf("marshaling queue snapshot: %v", err) + return + } + } + }) + } + + time.Sleep(500 * time.Millisecond) + close(stop) + wg.Wait() +} From 2ba4b6fdfbacd0132b6063ab0a836ca09fa03c4f Mon Sep 17 00:00:00 2001 From: lanfix Date: Thu, 10 Sep 2026 12:36:15 +0300 Subject: [PATCH 2/3] :bug: priorityqueue: Changes after review; --- pkg/controller/priorityqueue/priorityqueue.go | 10 +-- .../priorityqueue/priorityqueue_test.go | 68 ++++++++++++++++--- 2 files changed, 61 insertions(+), 17 deletions(-) diff --git a/pkg/controller/priorityqueue/priorityqueue.go b/pkg/controller/priorityqueue/priorityqueue.go index 1d81bd0aca..78353aabab 100644 --- a/pkg/controller/priorityqueue/priorityqueue.go +++ b/pkg/controller/priorityqueue/priorityqueue.go @@ -542,14 +542,8 @@ func (w *priorityqueue[T]) logState() { } } -// cloneItems returns a by-value deep copy of every queued item that is safe to -// use after w.lock is released. logState serializes the result outside the lock -// while other goroutines keep mutating the live items (handleWaitingItems sets -// ReadyAt to nil as an item becomes ready, lockedAddWithOpts reassigns it on the -// update path), so handing the *item[T] pointers straight to the logger is a -// data race. The race can crash the process from within encoding/json when a -// ReadyAt pointer is observed non-nil and then dereferenced after it went nil: -// "value method time.Time.MarshalJSON called using nil *Time pointer". +// cloneItems returns a deep copy of all queued items, taken under the lock, so that callers +// like logState can use them after it is released without racing with writers of ReadyAt. func (w *priorityqueue[T]) cloneItems() []item[T] { w.lock.Lock() defer w.lock.Unlock() diff --git a/pkg/controller/priorityqueue/priorityqueue_test.go b/pkg/controller/priorityqueue/priorityqueue_test.go index f6547443cc..0c4108a584 100644 --- a/pkg/controller/priorityqueue/priorityqueue_test.go +++ b/pkg/controller/priorityqueue/priorityqueue_test.go @@ -19,19 +19,24 @@ package priorityqueue import ( "encoding/json" "fmt" + "math" "math/rand/v2" + "reflect" "strconv" "sync" "testing" "testing/synctest" "time" + "github.com/google/go-cmp/cmp" fuzz "github.com/google/gofuzz" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/client-go/util/workqueue" "k8s.io/utils/ptr" + "k8s.io/utils/third_party/forked/golang/btree" + "sigs.k8s.io/randfill" ) var _ = Describe("Controllerworkqueue", func() { @@ -1070,15 +1075,9 @@ func TestPriorityQueueDoesNotDeadlockOnShutdownWhileHandingOutItems(t *testing.T } } -// TestPriorityQueueLogStateDoesNotRaceOnReadyAt is a regression test for a data -// race in logState. It used to collect the []*item[T] pointers under w.lock and -// then hand them to the logger after releasing the lock, while handleWaitingItems -// (toMove.ReadyAt = nil) and lockedAddWithOpts (item.ReadyAt = readyAt) kept -// mutating those same items. Serializing a ReadyAt pointer that races from -// non-nil to nil crashes the process inside encoding/json with -// "value method time.Time.MarshalJSON called using nil *Time pointer". -// cloneItems now returns an independent by-value copy taken under the lock; this -// test reproduces the concurrent access and must be run with -race. +// TestPriorityQueueLogStateDoesNotRaceOnReadyAt checks that snapshotting the queue is safe +// while other goroutines mutate ReadyAt. logState used to log the live *item[T] pointers, +// which raced and could crash encoding/json on a ReadyAt that concurrently became nil. func TestPriorityQueueLogStateDoesNotRaceOnReadyAt(t *testing.T) { t.Parallel() @@ -1128,3 +1127,54 @@ func TestPriorityQueueLogStateDoesNotRaceOnReadyAt(t *testing.T) { close(stop) wg.Wait() } + +// TestPriorityQueueCloneItemsCopiesAllFields checks that cloneItems returns items that are +// equal to the queued ones and share no memory with them. Filling the items with random +// values keeps both checks valid for fields that are added to item in the future. +func TestPriorityQueueCloneItemsCopiesAllFields(t *testing.T) { + t.Parallel() + + seed := time.Now().UnixNano() + t.Logf("seed: %d", seed) + filler := randfill.NewWithSeed(seed).NilChance(0).Funcs(func(t *time.Time, c randfill.Continue) { + *t = time.Unix(c.Int63n(math.MaxInt32), c.Int63n(int64(time.Second))) + }) + + for range 100 { + waitingItem, readyItem := &item[string]{}, &item[string]{} + filler.Fill(waitingItem) + filler.Fill(readyItem) + // Only items in the waiting tree have a ReadyAt. + readyItem.ReadyAt = nil + + q := &priorityqueue[string]{ + items: map[string]*item[string]{waitingItem.Key: waitingItem, readyItem.Key: readyItem}, + ready: btree.New(32, lessReady[string]), + waiting: btree.New(32, lessWaiting[string]), + } + q.waiting.ReplaceOrInsert(waitingItem) + q.ready.ReplaceOrInsert(readyItem) + + // cloneItems returns the waiting items first, then the ready ones. + expected := []item[string]{*waitingItem, *readyItem} + clones := q.cloneItems() + if diff := cmp.Diff(expected, clones); diff != "" { + t.Fatalf("cloned items don't match the queued ones: %s", diff) + } + + // Equal fields are not enough, a clone must not reference the item it was cloned from. + for i, original := range []*item[string]{waitingItem, readyItem} { + originalValue, cloneValue := reflect.ValueOf(*original), reflect.ValueOf(clones[i]) + for f := range cloneValue.NumField() { + field := cloneValue.Field(f) + if kind := field.Kind(); kind != reflect.Pointer && kind != reflect.Map && kind != reflect.Slice { + continue + } + + if !field.IsNil() && field.Pointer() == originalValue.Field(f).Pointer() { + t.Fatalf("field %s of the clone is shared with the queued item", cloneValue.Type().Field(f).Name) + } + } + } + } +} From c8b571908e9fa442ae158664a2d8fd936e129727 Mon Sep 17 00:00:00 2001 From: lanfix Date: Fri, 11 Sep 2026 11:04:58 +0300 Subject: [PATCH 3/3] :bug: priorityqueue: Changes after review, part 2; --- pkg/controller/priorityqueue/priorityqueue.go | 15 +- .../priorityqueue/priorityqueue_test.go | 138 +++++++----------- 2 files changed, 60 insertions(+), 93 deletions(-) diff --git a/pkg/controller/priorityqueue/priorityqueue.go b/pkg/controller/priorityqueue/priorityqueue.go index 78353aabab..aa9ab44e22 100644 --- a/pkg/controller/priorityqueue/priorityqueue.go +++ b/pkg/controller/priorityqueue/priorityqueue.go @@ -550,11 +550,7 @@ func (w *priorityqueue[T]) cloneItems() []item[T] { items := make([]item[T], 0, len(w.items)) appendItem := func(it *item[T]) bool { - clone := *it - if it.ReadyAt != nil { - clone.ReadyAt = new(*it.ReadyAt) - } - items = append(items, clone) + items = append(items, it.clone()) return true } w.waiting.Ascend(appendItem) @@ -584,6 +580,15 @@ type item[T comparable] struct { ReadyAt *time.Time `json:"readyAt,omitempty"` } +// clone returns a copy of the item that shares no memory with it. +func (i *item[T]) clone() item[T] { + clone := *i + if i.ReadyAt != nil { + clone.ReadyAt = new(*i.ReadyAt) + } + return clone +} + func (w *priorityqueue[T]) updateUnfinishedWorkLoop() { t := time.Tick(500 * time.Millisecond) // borrowed from workqueue: https://github.com/kubernetes/kubernetes/blob/67a807bf142c7a2a5ecfdb2a5d24b4cdea4cc79c/staging/src/k8s.io/client-go/util/workqueue/queue.go#L182 for { diff --git a/pkg/controller/priorityqueue/priorityqueue_test.go b/pkg/controller/priorityqueue/priorityqueue_test.go index 0c4108a584..798c671f6f 100644 --- a/pkg/controller/priorityqueue/priorityqueue_test.go +++ b/pkg/controller/priorityqueue/priorityqueue_test.go @@ -17,7 +17,6 @@ limitations under the License. package priorityqueue import ( - "encoding/json" "fmt" "math" "math/rand/v2" @@ -28,6 +27,7 @@ import ( "testing/synctest" "time" + "github.com/go-logr/logr/funcr" "github.com/google/go-cmp/cmp" fuzz "github.com/google/gofuzz" . "github.com/onsi/ginkgo/v2" @@ -35,7 +35,6 @@ import ( "k8s.io/apimachinery/pkg/util/sets" "k8s.io/client-go/util/workqueue" "k8s.io/utils/ptr" - "k8s.io/utils/third_party/forked/golang/btree" "sigs.k8s.io/randfill" ) @@ -620,11 +619,11 @@ func TestHighPriorityItemsAreReturnedBeforeLowPriorityItemMultipleTimes(t *testi } } -func newQueue() (*priorityqueue[string], *fakeMetricsProvider) { +func newQueue(opts ...Opt[string]) (*priorityqueue[string], *fakeMetricsProvider) { metrics := newFakeMetricsProvider() - q := New("test", func(o *Opts[string]) { + q := New("test", append([]Opt[string]{func(o *Opts[string]) { o.MetricProvider = metrics - }) + }}, opts...)...) q.(*priorityqueue[string]).waiting = &btreeInteractionValidator{ bTree: q.(*priorityqueue[string]).waiting, shouldHaveReadyAt: true, @@ -1075,106 +1074,69 @@ func TestPriorityQueueDoesNotDeadlockOnShutdownWhileHandingOutItems(t *testing.T } } -// TestPriorityQueueLogStateDoesNotRaceOnReadyAt checks that snapshotting the queue is safe -// while other goroutines mutate ReadyAt. logState used to log the live *item[T] pointers, -// which raced and could crash encoding/json on a ReadyAt that concurrently became nil. func TestPriorityQueueLogStateDoesNotRaceOnReadyAt(t *testing.T) { t.Parallel() + synctest.Test(t, func(t *testing.T) { + q, _ := newQueue(func(o *Opts[string]) { + o.Log = funcr.New(func(string, string) {}, funcr.Options{Verbosity: 5}) + }) + defer q.ShutDown() - q, _ := newQueue() - defer q.ShutDown() + // Re-adding an item with a lower After updates its ReadyAt, and every + // sleep lets logState log the item once, concurrently to the update. + for i := range 10 { + q.AddWithOpts(AddOpts{After: time.Hour - time.Duration(i)*time.Minute}, "foo") + time.Sleep(10 * time.Second) + } + }) +} - const keySpace = 20 - stop := make(chan struct{}) - var wg sync.WaitGroup +func TestItemCloneCopiesAllFields(t *testing.T) { + t.Parallel() - // Writers: churn items through waiting -> ready (handleWaitingItems clears - // ReadyAt) and through the update path (lockedAddWithOpts reassigns ReadyAt). - for range 8 { - wg.Go(func() { - for { - select { - case <-stop: - return - default: - } - q.AddWithOpts(AddOpts{ - After: time.Duration(rand.IntN(3)) * time.Millisecond, - Priority: new(rand.IntN(5)), - }, strconv.Itoa(rand.IntN(keySpace))) - } - }) - } + seed := time.Now().UnixNano() + t.Logf("seed: %d", seed) + filler := newItemFiller(seed) - // Readers: do what logState does - snapshot the queue and serialize it. - for range 4 { - wg.Go(func() { - for { - select { - case <-stop: - return - default: - } - if _, err := json.Marshal(q.cloneItems()); err != nil { - t.Errorf("marshaling queue snapshot: %v", err) - return - } - } - }) - } + for range 100 { + original := item[string]{} + filler.Fill(&original) - time.Sleep(500 * time.Millisecond) - close(stop) - wg.Wait() + if diff := cmp.Diff(original, original.clone()); diff != "" { + t.Errorf("clone doesn't match the original item: %s", diff) + } + } } -// TestPriorityQueueCloneItemsCopiesAllFields checks that cloneItems returns items that are -// equal to the queued ones and share no memory with them. Filling the items with random -// values keeps both checks valid for fields that are added to item in the future. -func TestPriorityQueueCloneItemsCopiesAllFields(t *testing.T) { +func TestItemCloneDoesNotShareMemoryWithOriginal(t *testing.T) { t.Parallel() seed := time.Now().UnixNano() t.Logf("seed: %d", seed) - filler := randfill.NewWithSeed(seed).NilChance(0).Funcs(func(t *time.Time, c randfill.Continue) { - *t = time.Unix(c.Int63n(math.MaxInt32), c.Int63n(int64(time.Second))) - }) + filler := newItemFiller(seed) for range 100 { - waitingItem, readyItem := &item[string]{}, &item[string]{} - filler.Fill(waitingItem) - filler.Fill(readyItem) - // Only items in the waiting tree have a ReadyAt. - readyItem.ReadyAt = nil - - q := &priorityqueue[string]{ - items: map[string]*item[string]{waitingItem.Key: waitingItem, readyItem.Key: readyItem}, - ready: btree.New(32, lessReady[string]), - waiting: btree.New(32, lessWaiting[string]), - } - q.waiting.ReplaceOrInsert(waitingItem) - q.ready.ReplaceOrInsert(readyItem) - - // cloneItems returns the waiting items first, then the ready ones. - expected := []item[string]{*waitingItem, *readyItem} - clones := q.cloneItems() - if diff := cmp.Diff(expected, clones); diff != "" { - t.Fatalf("cloned items don't match the queued ones: %s", diff) - } - - // Equal fields are not enough, a clone must not reference the item it was cloned from. - for i, original := range []*item[string]{waitingItem, readyItem} { - originalValue, cloneValue := reflect.ValueOf(*original), reflect.ValueOf(clones[i]) - for f := range cloneValue.NumField() { - field := cloneValue.Field(f) - if kind := field.Kind(); kind != reflect.Pointer && kind != reflect.Map && kind != reflect.Slice { - continue - } + original := item[string]{} + filler.Fill(&original) + + originalValue, cloneValue := reflect.ValueOf(original), reflect.ValueOf(original.clone()) + for i := range cloneValue.NumField() { + field := cloneValue.Field(i) + if kind := field.Kind(); kind != reflect.Pointer && kind != reflect.Map && kind != reflect.Slice { + continue + } - if !field.IsNil() && field.Pointer() == originalValue.Field(f).Pointer() { - t.Fatalf("field %s of the clone is shared with the queued item", cloneValue.Type().Field(f).Name) - } + if !field.IsNil() && field.Pointer() == originalValue.Field(i).Pointer() { + t.Fatalf("field %s of the clone is shared with the original item", cloneValue.Type().Field(i).Name) } } } } + +// newItemFiller returns a filler that also fills time.Time, which randfill +// leaves at its zero value as all of its fields are unexported. +func newItemFiller(seed int64) *randfill.Filler { + return randfill.NewWithSeed(seed).NilChance(0).Funcs(func(t *time.Time, c randfill.Continue) { + *t = time.Unix(c.Int63n(math.MaxInt32), c.Int63n(int64(time.Second))) + }) +}