Repository: kubernetes-sigs/controller-runtime

Issue:
Get from priority queue might get stuck when the priority queue is shut down
Hi team,

Recently I've testing with the new priority queue implementation and it occurs to me that right now getting from the priority queue when it has been shutdown might get stuck. It can be easily triggered if we:

1. create a priority queue
2. add an item to the queue
3. call the `Shutdown()` method
4. get from the priority queue
5. get again from the priority queue -> this call will be blocked

This happens as when the priority queue is shutdown, [the `spin()` method](https://github.com/kubernetes-sigs/controller-runtime/blob/b9a9ca01fd37a66225eb4793544349ee2ed1f196/pkg/controller/priorityqueue/priorityqueue.go#L198) will stop, but the `get` channel is [never closed](https://github.com/kubernetes-sigs/controller-runtime/blob/b9a9ca01fd37a66225eb4793544349ee2ed1f196/pkg/controller/priorityqueue/priorityqueue.go#L310), and [the procedure (step 5) that gets from the queue](https://github.com/kubernetes-sigs/controller-runtime/blob/b9a9ca01fd37a66225eb4793544349ee2ed1f196/pkg/controller/priorityqueue/priorityqueue.go#L276) will be stuck waiting on the `get` channel.

On the contrary, the rate limiting queue from the `workqueue` package will properly handle this situation properly by directly returning that the queue has been shut down.

The issue concerns me a bit as it will block graceful shutdown: in my project we use the `envtest` package and normally when the emulated cluster environment is stripped down, controller runtime will see that the main context has been cancelled and all reconcilers will stop upon seeing that the workqueue has been turned off; however, when priority queue is in use, the shutdown of the emulation will time out as there might be workers being stuck waiting for an item to arrive from the priority queue.

Attached also is some really simple test code that I used to trigger the behavior (with v0.21.0):

```go
import (
	"fmt"
	"testing"

	"k8s.io/client-go/util/workqueue"
	"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
)

func TestBlocker(t *testing.T) {
	rateLimiter := workqueue.DefaultTypedControllerRateLimiter[any]()
	wq := workqueue.NewTypedRateLimitingQueue(rateLimiter)

	println("Writing to the workqueue\n")
	wq.Add(1)

	println("Shutting down the workqueue\n")
	wq.ShutDown()

	println("Attempting to read from the workqueue (#1)\n")
        // This returns 1 and false.
	v, isShutDown := wq.Get()
	println(fmt.Sprintf("Value: %v, isShutDown: %v", v, isShutDown))

	println("Attempting to read from the workqueue (#2)\n")
        // This returns nil and true.
	v, isShutDown = wq.Get()
	println(fmt.Sprintf("Value: %v, isShutDown: %v", v, isShutDown))

	pq := priorityqueue.New[int]("testing")

	println("Writing to the priority queue\n")
	pq.Add(1)

	println("Shutting down the priority queue\n")
	pq.ShutDown()

	println("Attempting to read from the priority queue (#1)\n")
        // This returns 1, 0, and true.
	v, pri, isShutDown := pq.GetWithPriority()
	println(fmt.Sprintf("Value: %v, Priority: %v, isShutDown: %v", v, pri, isShutDown))

	println("Attempting to read from the priority queue (#2)\n")
        // This will never return.
	v, pri, isShutDown = pq.GetWithPriority()
	println(fmt.Sprintf("Value: %v, Priority: %v, isShutDown: %v", v, pri, isShutDown))
}
```

If this is already an known issue or if I misunderstanding anything, please let me know. Thanks.

Discussion:
@alvaroaleman: That sounds like a bug, please feel free to submit a patch.
/kind bug

Solution patch:
diff --git a/pkg/controller/priorityqueue/priorityqueue.go b/pkg/controller/priorityqueue/priorityqueue.go
index c3f77a6f..3f18918d 100644
--- a/pkg/controller/priorityqueue/priorityqueue.go
+++ b/pkg/controller/priorityqueue/priorityqueue.go
@@ -277,6 +277,13 @@ func (w *priorityqueue[T]) GetWithPriority() (_ T, priority int, shutdown bool)
 	w.waiters.Add(1)
 
 	w.notifyItemOrWaiterAdded()
+
+	// ref: https://github.com/kubernetes-sigs/controller-runtime/issues/3239
+	if w.shutdown.Load() {
+		var zero T
+		return zero, 0, true
+	}
+
 	item := <-w.get
 
 	return item.Key, item.Priority, w.shutdown.Load()



Test patch:
diff --git a/pkg/controller/priorityqueue/priorityqueue_test.go b/pkg/controller/priorityqueue/priorityqueue_test.go
index ec0f36e9..7e5d3ba3 100644
--- a/pkg/controller/priorityqueue/priorityqueue_test.go
+++ b/pkg/controller/priorityqueue/priorityqueue_test.go
@@ -300,6 +300,26 @@ var _ = Describe("Controllerworkqueue", func() {
 		Expect(metrics.depth["test"]).To(Equal(map[int]int{0: 2}))
 	})
 
+	// ref: https://github.com/kubernetes-sigs/controller-runtime/issues/3239
+	It("Get from priority queue might get stuck when the priority queue is shut down", func() {
+		q, _ := newQueue()
+
+		q.Add("baz")
+		// shut down
+		q.ShutDown()
+		q.AddWithOpts(AddOpts{After: time.Second}, "foo")
+
+		item, priority, isShutDown := q.GetWithPriority()
+		Expect(item).To(Equal(""))
+		Expect(priority).To(Equal(0))
+		Expect(isShutDown).To(BeTrue())
+
+		item1, priority1, isShutDown := q.GetWithPriority()
+		Expect(item1).To(Equal(""))
+		Expect(priority1).To(Equal(0))
+		Expect(isShutDown).To(BeTrue())
+	})
+
 	It("items are included in Len() and the queueDepth metric once they are ready", func() {
 		q, metrics := newQueue()
 		defer q.ShutDown()