Skip to content

Commit dc07eba

Browse files
ThomasK33ThomasK33
committed
fix: make watch broadcasts async and accept resourceVersion
1 parent 7738fe2 commit dc07eba

4 files changed

Lines changed: 70 additions & 25 deletions

File tree

‎internal/aggregated/storage/template.go‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -366,8 +366,7 @@ func (s *TemplateStorage) Create(
366366
return nil, fmt.Errorf("assertion failed: converted template must not be nil")
367367
}
368368

369-
//nolint:errcheck // Best-effort watch event broadcast.
370-
_ = s.broadcaster.Action(watch.Added, result.DeepCopy())
369+
broadcastEventAsync(s.broadcaster, watch.Added, result.DeepCopy())
371370

372371
return result, nil
373372
}
@@ -387,8 +386,7 @@ func (s *TemplateStorage) Create(
387386
return nil, fmt.Errorf("assertion failed: converted template must not be nil")
388387
}
389388

390-
//nolint:errcheck // Best-effort watch event broadcast.
391-
_ = s.broadcaster.Action(watch.Added, result.DeepCopy())
389+
broadcastEventAsync(s.broadcaster, watch.Added, result.DeepCopy())
392390

393391
return result, nil
394392
}
@@ -635,8 +633,7 @@ func (s *TemplateStorage) Update(
635633
return nil, false, fmt.Errorf("assertion failed: refreshed template must not be nil")
636634
}
637635

638-
//nolint:errcheck // Best-effort watch event broadcast.
639-
_ = s.broadcaster.Action(watch.Modified, result.DeepCopy())
636+
broadcastEventAsync(s.broadcaster, watch.Modified, result.DeepCopy())
640637

641638
return result, false, nil
642639
}
@@ -702,8 +699,7 @@ func (s *TemplateStorage) Delete(
702699
}
703700

704701
// Emit a Deleted event with the last-known template state.
705-
//nolint:errcheck // Best-effort watch event broadcast.
706-
_ = s.broadcaster.Action(watch.Deleted, templateObj.DeepCopy())
702+
broadcastEventAsync(s.broadcaster, watch.Deleted, templateObj.DeepCopy())
707703

708704
return &metav1.Status{Status: metav1.StatusSuccess}, true, nil
709705
}

‎internal/aggregated/storage/watch.go‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
88
"k8s.io/apimachinery/pkg/fields"
99
"k8s.io/apimachinery/pkg/labels"
10+
"k8s.io/apimachinery/pkg/runtime"
1011
"k8s.io/apimachinery/pkg/watch"
1112
)
1213

@@ -45,9 +46,6 @@ func validateUnsupportedWatchListOptions(opts *metainternalversion.ListOptions)
4546
return nil
4647
}
4748

48-
if opts.ResourceVersion != "" {
49-
return fmt.Errorf("resourceVersion %q is not supported for this watch endpoint", opts.ResourceVersion)
50-
}
5149
if opts.SendInitialEvents != nil && *opts.SendInitialEvents {
5250
return fmt.Errorf("sendInitialEvents=true is not supported for this watch endpoint")
5351
}
@@ -61,6 +59,21 @@ func validateUnsupportedWatchListOptions(opts *metainternalversion.ListOptions)
6159
return nil
6260
}
6361

62+
// broadcastEventAsync emits a watch event in a goroutine so mutation request
63+
// handlers are not blocked by slow watchers.
64+
func broadcastEventAsync(broadcaster *watch.Broadcaster, action watch.EventType, obj runtime.Object) {
65+
if broadcaster == nil {
66+
panic("assertion failed: watch broadcaster must not be nil")
67+
}
68+
if obj == nil {
69+
panic("assertion failed: watch event object must not be nil")
70+
}
71+
72+
go func() {
73+
_ = broadcaster.Action(action, obj)
74+
}()
75+
}
76+
6477
// filterForListOptions builds a watch.FilterFunc that applies namespace, label, and field selector filtering.
6578
// Returns nil if no filtering is needed.
6679
func filterForListOptions(requestNamespace string, opts *metainternalversion.ListOptions) (watch.FilterFunc, error) {

‎internal/aggregated/storage/watch_test.go‎

Lines changed: 47 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -308,7 +308,24 @@ func TestWorkspaceWatchRejectsUnsupportedWatchListOptions(t *testing.T) {
308308

309309
t.Run("resourceVersion", func(t *testing.T) {
310310
watcher, err := workspaceStorage.Watch(ctx, &metainternalversion.ListOptions{ResourceVersion: "123"})
311-
assertBadRequestWatchOptionsError(t, watcher, err, "resourceVersion")
311+
if err != nil {
312+
t.Fatalf("expected resourceVersion watch to be accepted, got %v", err)
313+
}
314+
if watcher == nil {
315+
t.Fatal("assertion failed: watcher must not be nil")
316+
}
317+
watcher.Stop()
318+
})
319+
320+
t.Run("legacyResourceVersionZero", func(t *testing.T) {
321+
watcher, err := workspaceStorage.Watch(ctx, &metainternalversion.ListOptions{ResourceVersion: "0"})
322+
if err != nil {
323+
t.Fatalf("expected resourceVersion=0 watch to be accepted, got %v", err)
324+
}
325+
if watcher == nil {
326+
t.Fatal("assertion failed: watcher must not be nil")
327+
}
328+
watcher.Stop()
312329
})
313330

314331
t.Run("resourceVersionMatch", func(t *testing.T) {
@@ -338,7 +355,24 @@ func TestTemplateWatchRejectsUnsupportedWatchListOptions(t *testing.T) {
338355

339356
t.Run("resourceVersion", func(t *testing.T) {
340357
watcher, err := templateStorage.Watch(ctx, &metainternalversion.ListOptions{ResourceVersion: "123"})
341-
assertBadRequestWatchOptionsError(t, watcher, err, "resourceVersion")
358+
if err != nil {
359+
t.Fatalf("expected resourceVersion watch to be accepted, got %v", err)
360+
}
361+
if watcher == nil {
362+
t.Fatal("assertion failed: watcher must not be nil")
363+
}
364+
watcher.Stop()
365+
})
366+
367+
t.Run("legacyResourceVersionZero", func(t *testing.T) {
368+
watcher, err := templateStorage.Watch(ctx, &metainternalversion.ListOptions{ResourceVersion: "0"})
369+
if err != nil {
370+
t.Fatalf("expected resourceVersion=0 watch to be accepted, got %v", err)
371+
}
372+
if watcher == nil {
373+
t.Fatal("assertion failed: watcher must not be nil")
374+
}
375+
watcher.Stop()
342376
})
343377

344378
t.Run("resourceVersionMatch", func(t *testing.T) {
@@ -381,9 +415,9 @@ func TestWatchRejectsDefaultedLegacyWatchListOptions(t *testing.T) {
381415
ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan,
382416
}
383417
workspaceWatcher, err = workspaceStorage.Watch(ctx, legacyOptionsWithZeroRV)
384-
assertBadRequestWatchOptionsError(t, workspaceWatcher, err, "resourceVersion")
418+
assertBadRequestWatchOptionsError(t, workspaceWatcher, err, "sendInitialEvents")
385419
templateWatcher, err = templateStorage.Watch(ctx, legacyOptionsWithZeroRV)
386-
assertBadRequestWatchOptionsError(t, templateWatcher, err, "resourceVersion")
420+
assertBadRequestWatchOptionsError(t, templateWatcher, err, "sendInitialEvents")
387421
}
388422

389423
func TestValidateUnsupportedWatchListOptions(t *testing.T) {
@@ -427,8 +461,8 @@ func TestValidateUnsupportedWatchListOptions(t *testing.T) {
427461
if err == nil {
428462
t.Fatal("expected defaulted legacy watch-list options with RV=0 to be rejected")
429463
}
430-
if !strings.Contains(err.Error(), "resourceVersion") {
431-
t.Fatalf("expected defaulted legacy watch-list RV=0 rejection to reference resourceVersion, got %v", err)
464+
if !strings.Contains(err.Error(), "sendInitialEvents") {
465+
t.Fatalf("expected defaulted legacy watch-list RV=0 rejection to reference sendInitialEvents, got %v", err)
432466
}
433467

434468
err = validateUnsupportedWatchListOptions(&metainternalversion.ListOptions{SendInitialEvents: &sendInitialEvents})
@@ -439,6 +473,11 @@ func TestValidateUnsupportedWatchListOptions(t *testing.T) {
439473
t.Fatalf("expected sendInitialEvents error, got %v", err)
440474
}
441475

476+
err = validateUnsupportedWatchListOptions(&metainternalversion.ListOptions{ResourceVersion: "123"})
477+
if err != nil {
478+
t.Fatalf("expected resourceVersion to be accepted, got %v", err)
479+
}
480+
442481
err = validateUnsupportedWatchListOptions(&metainternalversion.ListOptions{
443482
ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan,
444483
})
@@ -458,8 +497,8 @@ func TestValidateUnsupportedWatchListOptions(t *testing.T) {
458497
if err == nil {
459498
t.Fatal("expected non-legacy watch-list options to be rejected")
460499
}
461-
if !strings.Contains(err.Error(), "resourceVersion") {
462-
t.Fatalf("expected non-legacy watch-list rejection to reference resourceVersion, got %v", err)
500+
if !strings.Contains(err.Error(), "sendInitialEvents") {
501+
t.Fatalf("expected non-legacy watch-list rejection to reference sendInitialEvents, got %v", err)
463502
}
464503
}
465504

‎internal/aggregated/storage/workspace.go‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -368,8 +368,7 @@ func (s *WorkspaceStorage) Create(
368368
return nil, fmt.Errorf("assertion failed: converted workspace must not be nil")
369369
}
370370

371-
//nolint:errcheck // Best-effort watch event broadcast.
372-
_ = s.broadcaster.Action(watch.Added, result.DeepCopy())
371+
broadcastEventAsync(s.broadcaster, watch.Added, result.DeepCopy())
373372

374373
return result, nil
375374
}
@@ -519,8 +518,7 @@ func (s *WorkspaceStorage) Update(
519518
return nil, false, fmt.Errorf("assertion failed: converted workspace must not be nil")
520519
}
521520

522-
//nolint:errcheck // Best-effort watch event broadcast.
523-
_ = s.broadcaster.Action(watch.Modified, result.DeepCopy())
521+
broadcastEventAsync(s.broadcaster, watch.Modified, result.DeepCopy())
524522

525523
return result, false, nil
526524
}
@@ -595,8 +593,7 @@ func (s *WorkspaceStorage) Delete(
595593

596594
// Workspace deletion is asynchronous in Coder. Emit a Modified event
597595
// to signal that deletion was requested, rather than a Deleted event.
598-
//nolint:errcheck // Best-effort watch event broadcast.
599-
_ = s.broadcaster.Action(watch.Modified, workspaceObj.DeepCopy())
596+
broadcastEventAsync(s.broadcaster, watch.Modified, workspaceObj.DeepCopy())
600597

601598
// Deletion is asynchronous in Coder: we only enqueue a delete build transition here.
602599
// Report deleted=false so Kubernetes callers know the resource is not gone yet.

0 commit comments

Comments
 (0)