diff --git a/examples/priorityqueue/main.go b/examples/priorityqueue/main.go
index 2b09432f22..8dacdcc9a3 100644
--- a/examples/priorityqueue/main.go
+++ b/examples/priorityqueue/main.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"time"
 
+	"go.uber.org/zap/zapcore"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -45,7 +46,9 @@ func main() {
 }
 
 func run() error {
-	log.SetLogger(zap.New())
+	log.SetLogger(zap.New(func(o *zap.Options) {
+		o.Level = zapcore.Level(-5)
+	}))
 
 	// Setup a Manager
 	mgr, err := manager.New(kubeconfig.GetConfigOrDie(), manager.Options{
diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go
index ecffe07988..8f14bfdbfc 100644
--- a/pkg/cache/cache.go
+++ b/pkg/cache/cache.go
@@ -469,6 +469,8 @@ func defaultOpts(config *rest.Config, opts Options) (Options, error) {
 		}
 	}
 
+	opts.ByObject = maps.Clone(opts.ByObject)
+	opts.DefaultNamespaces = maps.Clone(opts.DefaultNamespaces)
 	for obj, byObject := range opts.ByObject {
 		isNamespaced, err := apiutil.IsObjectNamespaced(obj, opts.Scheme, opts.Mapper)
 		if err != nil {
@@ -480,6 +482,8 @@ func defaultOpts(config *rest.Config, opts Options) (Options, error) {
 
 		if isNamespaced && byObject.Namespaces == nil {
 			byObject.Namespaces = maps.Clone(opts.DefaultNamespaces)
+		} else {
+			byObject.Namespaces = maps.Clone(byObject.Namespaces)
 		}
 
 		// Default the namespace-level configs first, because they need to use the undefaulted type-level config
@@ -487,7 +491,6 @@ func defaultOpts(config *rest.Config, opts Options) (Options, error) {
 		for namespace, config := range byObject.Namespaces {
 			// 1. Default from the undefaulted type-level config
 			config = defaultConfig(config, byObjectToConfig(byObject))
-
 			// 2. Default from the namespace-level config. This was defaulted from the global default config earlier, but
 			//    might not have an entry for the current namespace.
 			if defaultNamespaceSettings, hasDefaultNamespace := opts.DefaultNamespaces[namespace]; hasDefaultNamespace {
diff --git a/pkg/cache/defaulting_test.go b/pkg/cache/defaulting_test.go
index 8e3033eb47..d9d0dcceb3 100644
--- a/pkg/cache/defaulting_test.go
+++ b/pkg/cache/defaulting_test.go
@@ -18,6 +18,7 @@ package cache
 
 import (
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 
@@ -432,6 +433,34 @@ func TestDefaultOpts(t *testing.T) {
 	}
 }
 
+func TestDefaultOptsRace(t *testing.T) {
+	opts := Options{
+		Mapper: &fakeRESTMapper{},
+		ByObject: map[client.Object]ByObject{
+			&corev1.Pod{}: {
+				Label: labels.SelectorFromSet(map[string]string{"from": "pod"}),
+				Namespaces: map[string]Config{"default": {
+					LabelSelector: labels.SelectorFromSet(map[string]string{"from": "pod"}),
+				}},
+			},
+		},
+		DefaultNamespaces: map[string]Config{"default": {}},
+	}
+
+	// Start go routines which re-use the above options struct.
+	wg := sync.WaitGroup{}
+	for range 2 {
+		wg.Add(1)
+		go func() {
+			_, _ = defaultOpts(&rest.Config{}, opts)
+			wg.Done()
+		}()
+	}
+
+	// Wait for the go routines to finish.
+	wg.Wait()
+}
+
 type fakeRESTMapper struct {
 	meta.RESTMapper
 }
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index b7d7286033..5c5b249ef5 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -202,6 +202,7 @@ func NewTypedUnmanaged[request comparable](name string, mgr manager.Manager, opt
 		options.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request] {
 			if ptr.Deref(mgr.GetControllerOptions().UsePriorityQueue, false) {
 				return priorityqueue.New(controllerName, func(o *priorityqueue.Opts[request]) {
+					o.Log = mgr.GetLogger().WithValues("controller", controllerName)
 					o.RateLimiter = rateLimiter
 				})
 			}
diff --git a/pkg/controller/priorityqueue/metrics.go b/pkg/controller/priorityqueue/metrics.go
index f6a2697a65..36626646f4 100644
--- a/pkg/controller/priorityqueue/metrics.go
+++ b/pkg/controller/priorityqueue/metrics.go
@@ -85,11 +85,11 @@ func (m *defaultQueueMetrics[T]) get(item T) {
 		return
 	}
 
+	m.depth.Dec()
+
 	m.mapLock.Lock()
 	defer m.mapLock.Unlock()
 
-	m.depth.Dec()
-
 	m.processingStartTimes[item] = m.clock.Now()
 	if startTime, exists := m.addTimes[item]; exists {
 		m.latency.Observe(m.sinceInSeconds(startTime))
diff --git a/pkg/controller/priorityqueue/priorityqueue.go b/pkg/controller/priorityqueue/priorityqueue.go
index 2b3a8904d7..996369f47a 100644
--- a/pkg/controller/priorityqueue/priorityqueue.go
+++ b/pkg/controller/priorityqueue/priorityqueue.go
@@ -5,6 +5,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/go-logr/logr"
 	"github.com/google/btree"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/util/workqueue"
@@ -36,6 +37,7 @@ type Opts[T comparable] struct {
 	// limiter with an initial delay of five milliseconds and a max delay of 1000 seconds.
 	RateLimiter    workqueue.TypedRateLimiter[T]
 	MetricProvider workqueue.MetricsProvider
+	Log            logr.Logger
 }
 
 // Opt allows to configure a PriorityQueue.
@@ -57,6 +59,7 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
 	}
 
 	pq := &priorityqueue[T]{
+		log:         opts.Log,
 		items:       map[T]*item[T]{},
 		queue:       btree.NewG(32, less[T]),
 		becameReady: sets.Set[T]{},
@@ -75,6 +78,7 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
 	}
 
 	go pq.spin()
+	go pq.logState()
 	if _, ok := pq.metrics.(noMetrics[T]); !ok {
 		go pq.updateUnfinishedWorkLoop()
 	}
@@ -83,6 +87,7 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
 }
 
 type priorityqueue[T comparable] struct {
+	log logr.Logger
 	// lock has to be acquired for any access any of items, queue, addedCounter
 	// or becameReady
 	lock  sync.Mutex
@@ -141,14 +146,14 @@ func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 		}
 		if _, ok := w.items[key]; !ok {
 			item := &item[T]{
-				key:          key,
-				addedCounter: w.addedCounter,
-				priority:     o.Priority,
-				readyAt:      readyAt,
+				Key:          key,
+				AddedCounter: w.addedCounter,
+				Priority:     o.Priority,
+				ReadyAt:      readyAt,
 			}
 			w.items[key] = item
 			w.queue.ReplaceOrInsert(item)
-			if item.readyAt == nil {
+			if item.ReadyAt == nil {
 				w.metrics.add(key)
 			}
 			w.addedCounter++
@@ -158,15 +163,15 @@ func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 		// The b-tree de-duplicates based on ordering and any change here
 		// will affect the order - Just delete and re-add.
 		item, _ := w.queue.Delete(w.items[key])
-		if o.Priority > item.priority {
-			item.priority = o.Priority
+		if o.Priority > item.Priority {
+			item.Priority = o.Priority
 		}
 
-		if item.readyAt != nil && (readyAt == nil || readyAt.Before(*item.readyAt)) {
-			if readyAt == nil {
+		if item.ReadyAt != nil && (readyAt == nil || readyAt.Before(*item.ReadyAt)) {
+			if readyAt == nil && !w.becameReady.Has(key) {
 				w.metrics.add(key)
 			}
-			item.readyAt = readyAt
+			item.ReadyAt = readyAt
 		}
 
 		w.queue.ReplaceOrInsert(item)
@@ -210,14 +215,14 @@ func (w *priorityqueue[T]) spin() {
 			// track what we want to delete and do it after we are done ascending.
 			var toDelete []*item[T]
 			w.queue.Ascend(func(item *item[T]) bool {
-				if item.readyAt != nil {
-					if readyAt := item.readyAt.Sub(w.now()); readyAt > 0 {
+				if item.ReadyAt != nil {
+					if readyAt := item.ReadyAt.Sub(w.now()); readyAt > 0 {
 						nextReady = w.tick(readyAt)
 						return false
 					}
-					if !w.becameReady.Has(item.key) {
-						w.metrics.add(item.key)
-						w.becameReady.Insert(item.key)
+					if !w.becameReady.Has(item.Key) {
+						w.metrics.add(item.Key)
+						w.becameReady.Insert(item.Key)
 					}
 				}
 
@@ -228,16 +233,16 @@ func (w *priorityqueue[T]) spin() {
 				}
 
 				// Item is locked, we can not hand it out
-				if w.locked.Has(item.key) {
+				if w.locked.Has(item.Key) {
 					return true
 				}
 
-				w.metrics.get(item.key)
-				w.locked.Insert(item.key)
+				w.metrics.get(item.Key)
+				w.locked.Insert(item.Key)
 				w.waiters.Add(-1)
-				delete(w.items, item.key)
+				delete(w.items, item.Key)
 				toDelete = append(toDelete, item)
-				w.becameReady.Delete(item.key)
+				w.becameReady.Delete(item.Key)
 				w.get <- *item
 
 				return true
@@ -268,7 +273,7 @@ func (w *priorityqueue[T]) GetWithPriority() (_ T, priority int, shutdown bool)
 	w.notifyItemOrWaiterAdded()
 	item := <-w.get
 
-	return item.key, item.priority, w.shutdown.Load()
+	return item.Key, item.Priority, w.shutdown.Load()
 }
 
 func (w *priorityqueue[T]) Get() (item T, shutdown bool) {
@@ -316,7 +321,7 @@ func (w *priorityqueue[T]) Len() int {
 
 	var result int
 	w.queue.Ascend(func(item *item[T]) bool {
-		if item.readyAt == nil || item.readyAt.Compare(w.now()) <= 0 {
+		if item.ReadyAt == nil || item.ReadyAt.Compare(w.now()) <= 0 {
 			result++
 			return true
 		}
@@ -326,36 +331,64 @@ func (w *priorityqueue[T]) Len() int {
 	return result
 }
 
+func (w *priorityqueue[T]) logState() {
+	t := time.Tick(10 * time.Second)
+	for {
+		select {
+		case <-w.done:
+			return
+		case <-t:
+		}
+
+		// Log level may change at runtime, so keep the
+		// loop going even if a given level is currently
+		// not enabled.
+		if !w.log.V(5).Enabled() {
+			continue
+		}
+		w.lock.Lock()
+		items := make([]*item[T], 0, len(w.items))
+		w.queue.Ascend(func(item *item[T]) bool {
+			items = append(items, item)
+			return true
+		})
+		w.lock.Unlock()
+
+		w.log.V(5).Info("workqueue_items", "items", items)
+	}
+}
+
 func less[T comparable](a, b *item[T]) bool {
-	if a.readyAt == nil && b.readyAt != nil {
+	if a.ReadyAt == nil && b.ReadyAt != nil {
 		return true
 	}
-	if b.readyAt == nil && a.readyAt != nil {
+	if b.ReadyAt == nil && a.ReadyAt != nil {
 		return false
 	}
-	if a.readyAt != nil && b.readyAt != nil && !a.readyAt.Equal(*b.readyAt) {
-		return a.readyAt.Before(*b.readyAt)
+	if a.ReadyAt != nil && b.ReadyAt != nil && !a.ReadyAt.Equal(*b.ReadyAt) {
+		return a.ReadyAt.Before(*b.ReadyAt)
 	}
-	if a.priority != b.priority {
-		return a.priority > b.priority
+	if a.Priority != b.Priority {
+		return a.Priority > b.Priority
 	}
 
-	return a.addedCounter < b.addedCounter
+	return a.AddedCounter < b.AddedCounter
 }
 
 type item[T comparable] struct {
-	key          T
-	addedCounter uint64
-	priority     int
-	readyAt      *time.Time
+	Key          T          `json:"key"`
+	AddedCounter uint64     `json:"addedCounter"`
+	Priority     int        `json:"priority"`
+	ReadyAt      *time.Time `json:"readyAt,omitempty"`
 }
 
 func (w *priorityqueue[T]) updateUnfinishedWorkLoop() {
-	t := time.NewTicker(500 * time.Millisecond) // borrowed from workqueue: https://github.com/kubernetes/kubernetes/blob/67a807bf142c7a2a5ecfdb2a5d24b4cdea4cc79c/staging/src/k8s.io/client-go/util/workqueue/queue.go#L182
-	defer t.Stop()
-	for range t.C {
-		if w.shutdown.Load() {
+	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 {
+		select {
+		case <-w.done:
 			return
+		case <-t:
 		}
 		w.metrics.updateUnfinishedWork()
 	}
diff --git a/pkg/controller/priorityqueue/priorityqueue_test.go b/pkg/controller/priorityqueue/priorityqueue_test.go
index e431c993fb..18de95aac2 100644
--- a/pkg/controller/priorityqueue/priorityqueue_test.go
+++ b/pkg/controller/priorityqueue/priorityqueue_test.go
@@ -395,6 +395,48 @@ var _ = Describe("Controllerworkqueue", func() {
 		Expect(q.Len()).To(Equal(1))
 		metrics.mu.Lock()
 		Expect(metrics.depth["test"]).To(Equal(1))
+		metrics.mu.Unlock()
+
+		// Get the item to ensure the codepath in
+		// `spin` for the metrics is passed by so
+		// that this starts failing if it incorrectly
+		// calls `metrics.add` again.
+		item, _ := q.Get()
+		Expect(item).To(Equal("foo"))
+		Expect(q.Len()).To(Equal(0))
+		metrics.mu.Lock()
+		Expect(metrics.depth["test"]).To(Equal(0))
+		metrics.mu.Unlock()
+	})
+
+	It("Updates metrics correctly for an item whose requeueAfter expired that gets added again without requeueAfter", func() {
+		q, metrics := newQueue()
+		defer q.ShutDown()
+
+		q.AddWithOpts(AddOpts{After: 50 * time.Millisecond}, "foo")
+		time.Sleep(100 * time.Millisecond)
+
+		Expect(q.Len()).To(Equal(1))
+		metrics.mu.Lock()
+		Expect(metrics.depth["test"]).To(Equal(1))
+		metrics.mu.Unlock()
+
+		q.AddWithOpts(AddOpts{}, "foo")
+		Expect(q.Len()).To(Equal(1))
+		metrics.mu.Lock()
+		Expect(metrics.depth["test"]).To(Equal(1))
+		metrics.mu.Unlock()
+
+		// Get the item to ensure the codepath in
+		// `spin` for the metrics is passed by so
+		// that this starts failing if it incorrectly
+		// calls `metrics.add` again.
+		item, _ := q.Get()
+		Expect(item).To(Equal("foo"))
+		Expect(q.Len()).To(Equal(0))
+		metrics.mu.Lock()
+		Expect(metrics.depth["test"]).To(Equal(0))
+		metrics.mu.Unlock()
 	})
 })
 
diff --git a/pkg/internal/controller/controller.go b/pkg/internal/controller/controller.go
index fda25e0641..cc734dfb61 100644
--- a/pkg/internal/controller/controller.go
+++ b/pkg/internal/controller/controller.go
@@ -175,7 +175,16 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 		// caches.
 		errGroup := &errgroup.Group{}
 		for _, watch := range c.startWatches {
-			log := c.LogConstructor(nil).WithValues("source", fmt.Sprintf("%s", watch))
+			log := c.LogConstructor(nil)
+			_, ok := watch.(interface {
+				String() string
+			})
+
+			if !ok {
+				log = log.WithValues("source", fmt.Sprintf("%T", watch))
+			} else {
+				log = log.WithValues("source", fmt.Sprintf("%s", watch))
+			}
 			didStartSyncingSource := &atomic.Bool{}
 			errGroup.Go(func() error {
 				// Use a timeout for starting and syncing the source to avoid silently
@@ -191,7 +200,7 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 						sourceStartErrChan <- err
 						return
 					}
-					syncingSource, ok := watch.(source.SyncingSource)
+					syncingSource, ok := watch.(source.TypedSyncingSource[request])
 					if !ok {
 						return
 					}
diff --git a/pkg/internal/controller/controller_test.go b/pkg/internal/controller/controller_test.go
index 52f45612f2..3a23156a9c 100644
--- a/pkg/internal/controller/controller_test.go
+++ b/pkg/internal/controller/controller_test.go
@@ -46,6 +46,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
+type TestRequest struct {
+	Key string
+}
+
 var _ = Describe("controller", func() {
 	var fakeReconcile *fakeReconciler
 	var ctrl *Controller[reconcile.Request]
@@ -340,6 +344,41 @@ var _ = Describe("controller", func() {
 			Expect(err.Error()).To(Equal("controller was started more than once. This is likely to be caused by being added to a manager multiple times"))
 		})
 
+		It("should check for correct TypedSyncingSource if custom types are used", func() {
+			queue := &controllertest.TypedQueue[TestRequest]{
+				TypedInterface: workqueue.NewTyped[TestRequest](),
+			}
+			ctrl := &Controller[TestRequest]{
+				NewQueue: func(string, workqueue.TypedRateLimiter[TestRequest]) workqueue.TypedRateLimitingInterface[TestRequest] {
+					return queue
+				},
+				LogConstructor: func(*TestRequest) logr.Logger {
+					return log.RuntimeLog.WithName("controller").WithName("test")
+				},
+			}
+			ctrl.CacheSyncTimeout = time.Second
+			src := &bisignallingSource[TestRequest]{
+				startCall: make(chan workqueue.TypedRateLimitingInterface[TestRequest]),
+				startDone: make(chan error, 1),
+				waitCall:  make(chan struct{}),
+				waitDone:  make(chan error, 1),
+			}
+			ctrl.startWatches = []source.TypedSource[TestRequest]{src}
+			ctrl.Name = "foo"
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			startCh := make(chan error)
+			go func() {
+				defer GinkgoRecover()
+				startCh <- ctrl.Start(ctx)
+			}()
+			Eventually(src.startCall).Should(Receive(Equal(queue)))
+			src.startDone <- nil
+			Eventually(src.waitCall).Should(BeClosed())
+			src.waitDone <- nil
+			cancel()
+			Eventually(startCh).Should(Receive(Succeed()))
+		})
 	})
 
 	Describe("Processing queue items from a Controller", func() {
@@ -901,3 +940,40 @@ func (c *cacheWithIndefinitelyBlockingGetInformer) GetInformer(ctx context.Conte
 	<-ctx.Done()
 	return nil, errors.New("GetInformer timed out")
 }
+
+type bisignallingSource[T comparable] struct {
+	// receives the queue that is passed to Start
+	startCall chan workqueue.TypedRateLimitingInterface[T]
+	// passes an error to return from Start
+	startDone chan error
+	// closed when WaitForSync is called
+	waitCall chan struct{}
+	// passes an error to return from WaitForSync
+	waitDone chan error
+}
+
+var _ source.TypedSyncingSource[int] = (*bisignallingSource[int])(nil)
+
+func (t *bisignallingSource[T]) Start(ctx context.Context, q workqueue.TypedRateLimitingInterface[T]) error {
+	select {
+	case t.startCall <- q:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	select {
+	case err := <-t.startDone:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (t *bisignallingSource[T]) WaitForSync(ctx context.Context) error {
+	close(t.waitCall)
+	select {
+	case err := <-t.waitDone:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}