fix: serialize watch event dispatch per storage

Author: ThomasK33

internal/aggregated/storage/template.go

@@ -64,6 +64,8 @@ type TemplateStorage struct {
 	provider       coder.ClientProvider
 	tableConvertor rest.TableConvertor
 	broadcaster    *watch.Broadcaster
+	watchEvents    chan watch.Event
+	watchEventsWG  sync.WaitGroup
 	destroyOnce    sync.Once
 }
 
@@ -73,11 +75,16 @@ func NewTemplateStorage(provider coder.ClientProvider) *TemplateStorage {
 		panic("assertion failed: template client provider must not be nil")
 	}
 
-	return &TemplateStorage{
+	storage := &TemplateStorage{
 		provider:       provider,
 		tableConvertor: rest.NewDefaultTableConvertor(aggregationv1alpha1.Resource("codertemplates")),
 		broadcaster:    watch.NewBroadcaster(watchBroadcasterQueueLen, watch.WaitIfChannelFull),
+		watchEvents:    make(chan watch.Event, watchBroadcasterQueueLen),
 	}
+	storage.watchEventsWG.Add(1)
+	go storage.dispatchWatchEvents()
+
+	return storage
 }
 
 // New returns an empty CoderTemplate object.
@@ -92,6 +99,12 @@ func (s *TemplateStorage) Destroy() {
 	}
 
 	s.destroyOnce.Do(func() {
+		if s.watchEvents == nil {
+			panic("assertion failed: template watch event queue must not be nil")
+		}
+		close(s.watchEvents)
+		s.watchEventsWG.Wait()
+
 		if s.broadcaster != nil {
 			s.broadcaster.Shutdown()
 		}
@@ -366,7 +379,7 @@ func (s *TemplateStorage) Create(
 			return nil, fmt.Errorf("assertion failed: converted template must not be nil")
 		}
 
-		broadcastEventAsync(s.broadcaster, watch.Added, result.DeepCopy())
+		s.enqueueWatchEvent(watch.Added, result.DeepCopy())
 
 		return result, nil
 	}
@@ -386,7 +399,7 @@ func (s *TemplateStorage) Create(
 		return nil, fmt.Errorf("assertion failed: converted template must not be nil")
 	}
 
-	broadcastEventAsync(s.broadcaster, watch.Added, result.DeepCopy())
+	s.enqueueWatchEvent(watch.Added, result.DeepCopy())
 
 	return result, nil
 }
@@ -633,7 +646,7 @@ func (s *TemplateStorage) Update(
 		return nil, false, fmt.Errorf("assertion failed: refreshed template must not be nil")
 	}
 
-	broadcastEventAsync(s.broadcaster, watch.Modified, result.DeepCopy())
+	s.enqueueWatchEvent(watch.Modified, result.DeepCopy())
 
 	return result, false, nil
 }
@@ -699,11 +712,33 @@ func (s *TemplateStorage) Delete(
 	}
 
 	// Emit a Deleted event with the last-known template state.
-	broadcastEventAsync(s.broadcaster, watch.Deleted, templateObj.DeepCopy())
+	s.enqueueWatchEvent(watch.Deleted, templateObj.DeepCopy())
 
 	return &metav1.Status{Status: metav1.StatusSuccess}, true, nil
 }
 
+func (s *TemplateStorage) dispatchWatchEvents() {
+	defer s.watchEventsWG.Done()
+
+	for event := range s.watchEvents {
+		_ = s.broadcaster.Action(event.Type, event.Object)
+	}
+}
+
+func (s *TemplateStorage) enqueueWatchEvent(action watch.EventType, obj runtime.Object) {
+	if s == nil {
+		panic("assertion failed: template storage must not be nil")
+	}
+	if s.watchEvents == nil {
+		panic("assertion failed: template watch event queue must not be nil")
+	}
+	if obj == nil {
+		panic("assertion failed: template watch event object must not be nil")
+	}
+
+	s.watchEvents <- watch.Event{Type: action, Object: obj}
+}
+
 // ConvertToTable converts a template object or list into kubectl table output.
 func (s *TemplateStorage) ConvertToTable(ctx context.Context, object, tableOptions runtime.Object) (*metav1.Table, error) {
 	if s == nil {

internal/aggregated/storage/watch.go

@@ -7,7 +7,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
@@ -59,21 +58,6 @@ func validateUnsupportedWatchListOptions(opts *metainternalversion.ListOptions)
 	return nil
 }
 
-// broadcastEventAsync emits a watch event in a goroutine so mutation request
-// handlers are not blocked by slow watchers.
-func broadcastEventAsync(broadcaster *watch.Broadcaster, action watch.EventType, obj runtime.Object) {
-	if broadcaster == nil {
-		panic("assertion failed: watch broadcaster must not be nil")
-	}
-	if obj == nil {
-		panic("assertion failed: watch event object must not be nil")
-	}
-
-	go func() {
-		_ = broadcaster.Action(action, obj)
-	}()
-}
-
 // filterForListOptions builds a watch.FilterFunc that applies namespace, label, and field selector filtering.
 // Returns nil if no filtering is needed.
 func filterForListOptions(requestNamespace string, opts *metainternalversion.ListOptions) (watch.FilterFunc, error) {

internal/aggregated/storage/workspace.go

@@ -38,6 +38,8 @@ type WorkspaceStorage struct {
 	provider       coder.ClientProvider
 	tableConvertor rest.TableConvertor
 	broadcaster    *watch.Broadcaster
+	watchEvents    chan watch.Event
+	watchEventsWG  sync.WaitGroup
 	destroyOnce    sync.Once
 }
 
@@ -47,11 +49,16 @@ func NewWorkspaceStorage(provider coder.ClientProvider) *WorkspaceStorage {
 		panic("assertion failed: workspace client provider must not be nil")
 	}
 
-	return &WorkspaceStorage{
+	storage := &WorkspaceStorage{
 		provider:       provider,
 		tableConvertor: rest.NewDefaultTableConvertor(aggregationv1alpha1.Resource("coderworkspaces")),
 		broadcaster:    watch.NewBroadcaster(watchBroadcasterQueueLen, watch.WaitIfChannelFull),
+		watchEvents:    make(chan watch.Event, watchBroadcasterQueueLen),
 	}
+	storage.watchEventsWG.Add(1)
+	go storage.dispatchWatchEvents()
+
+	return storage
 }
 
 // New returns an empty CoderWorkspace object.
@@ -66,6 +73,12 @@ func (s *WorkspaceStorage) Destroy() {
 	}
 
 	s.destroyOnce.Do(func() {
+		if s.watchEvents == nil {
+			panic("assertion failed: workspace watch event queue must not be nil")
+		}
+		close(s.watchEvents)
+		s.watchEventsWG.Wait()
+
 		if s.broadcaster != nil {
 			s.broadcaster.Shutdown()
 		}
@@ -368,7 +381,7 @@ func (s *WorkspaceStorage) Create(
 		return nil, fmt.Errorf("assertion failed: converted workspace must not be nil")
 	}
 
-	broadcastEventAsync(s.broadcaster, watch.Added, result.DeepCopy())
+	s.enqueueWatchEvent(watch.Added, result.DeepCopy())
 
 	return result, nil
 }
@@ -518,7 +531,7 @@ func (s *WorkspaceStorage) Update(
 		return nil, false, fmt.Errorf("assertion failed: converted workspace must not be nil")
 	}
 
-	broadcastEventAsync(s.broadcaster, watch.Modified, result.DeepCopy())
+	s.enqueueWatchEvent(watch.Modified, result.DeepCopy())
 
 	return result, false, nil
 }
@@ -593,13 +606,35 @@ func (s *WorkspaceStorage) Delete(
 
 	// Workspace deletion is asynchronous in Coder. Emit a Modified event
 	// to signal that deletion was requested, rather than a Deleted event.
-	broadcastEventAsync(s.broadcaster, watch.Modified, workspaceObj.DeepCopy())
+	s.enqueueWatchEvent(watch.Modified, workspaceObj.DeepCopy())
 
 	// Deletion is asynchronous in Coder: we only enqueue a delete build transition here.
 	// Report deleted=false so Kubernetes callers know the resource is not gone yet.
 	return &metav1.Status{Status: metav1.StatusSuccess}, false, nil
 }
 
+func (s *WorkspaceStorage) dispatchWatchEvents() {
+	defer s.watchEventsWG.Done()
+
+	for event := range s.watchEvents {
+		_ = s.broadcaster.Action(event.Type, event.Object)
+	}
+}
+
+func (s *WorkspaceStorage) enqueueWatchEvent(action watch.EventType, obj runtime.Object) {
+	if s == nil {
+		panic("assertion failed: workspace storage must not be nil")
+	}
+	if s.watchEvents == nil {
+		panic("assertion failed: workspace watch event queue must not be nil")
+	}
+	if obj == nil {
+		panic("assertion failed: workspace watch event object must not be nil")
+	}
+
+	s.watchEvents <- watch.Event{Type: action, Object: obj}
+}
+
 // ConvertToTable converts a workspace object or list into kubectl table output.
 func (s *WorkspaceStorage) ConvertToTable(ctx context.Context, object, tableOptions runtime.Object) (*metav1.Table, error) {
 	if s == nil {