Priority queue: priority not always respected #3363

I suspect that there may be conditions under which the item returned by GetWithPriority() is not always the correct/expected element.
Both GetWithPriority() and AddWithOpts trigger a spin():

In the GetWithPriority(), the w.waiters are raised w.waiters.Add(1), and in the spin items are skipped if none is waiting:

				if w.waiters.Load() == 0 {
					// Have to keep iterating here to ensure we update metrics
					// for further items that became ready and set nextReady.
					return true
				}

If a GetWithPriority() is executed right after an AddWithOpts, there could be the possibility that the Ascend of the Btree is not completed yet, causing not all the items of the BTree to evaluate the same w.waiters.Load() value.

This will result in the non-obtaining of the higher-priority element.

In particular, you can see that this test https://github.com/kubevirt/kubevirt/blob/66f09757e8707ae4a09c660119ec1ab3658b0cdb/pkg/virt-controller/watch/migration/migration_test.go#L2342 is flaky (a failure example: https://prow.ci.kubevirt.io/view/gs/kubevirt-prow/pr-logs/pull/kubevirt_kubevirt/15887/pull-kubevirt-unit-test/1981424912933326848).

A possible solution could be adding a lock before increasing the waiters:

might be:

 func (w *priorityqueue[T]) GetWithPriority() (_ T, priority int, shutdown bool) { 
 	if w.shutdown.Load() { 
 		var zero T 
 		return zero, 0, true 
 	} 
    w.lock.Lock()
 	w.waiters.Add(1) 
    w.lock.Unlock()

What do you think?
Thank you!

Edit: Checked the above solution. The flaky disappeared, but IDK the implications