From beae2a10a7eb0da0ac76e4492b33b7369a518f03 Mon Sep 17 00:00:00 2001 From: David Negstad Date: Mon, 19 Oct 2026 15:24:19 -0700 Subject: [PATCH] [release/0.26] Harden watcher consumption and adopt bounded Tilt watch delivery Backport PR #292 (06562f36aa9361e4f58db10626f9f9aac5cdb8b2) onto release/0.26 without merging main or unrelated release/0.27 feature work. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a9488cee-a143-4b72-b788-d9d0d2f7004e --- NOTICE | 4 +- api/v1/log_storage.go | 36 +-- api/v1/log_storage_test.go | 104 ++++++++ go.mod | 2 +- go.sum | 4 +- .../v2_namespace_lifecycle_storage_test.go | 46 ---- .../v2_namespace_lifecycle_watcher.go | 85 ++++--- .../v2_namespace_lifecycle_watcher_test.go | 228 +++++++++++++++++- 8 files changed, 394 insertions(+), 115 deletions(-) create mode 100644 api/v1/log_storage_test.go diff --git a/NOTICE b/NOTICE index e59afc5d..7d42d399 100644 --- a/NOTICE +++ b/NOTICE @@ -8405,8 +8405,8 @@ OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. ---------------------------------------------------------- -github.com/tilt-dev/tilt-apiserver/pkg v0.19.1 - Apache-2.0 -https://github.com/tilt-dev/tilt-apiserver/blob/v0.19.1/LICENSE +github.com/tilt-dev/tilt-apiserver/pkg v0.19.2 - Apache-2.0 +https://github.com/tilt-dev/tilt-apiserver/blob/v0.19.2/LICENSE Apache License Version 2.0, January 2004 diff --git a/api/v1/log_storage.go b/api/v1/log_storage.go index 2226cc7a..7f849c55 100644 --- a/api/v1/log_storage.go +++ b/api/v1/log_storage.go @@ -181,6 +181,7 @@ func NewLogStorage(parentKindStorage registry_rest.StandardStorage, logStreamer resourceWatchers: &syncmap.Map[types.UID, *syncmap.Map[*resourceWatcher, bool]]{}, mutex: &sync.Mutex{}, watcher: nil, + disposeCh: make(chan struct{}), }, nil } @@ -193,6 +194,13 @@ func (ls *LogStorage) watchResource(uid types.UID, log logr.Logger) (*resourceWa return nil, fmt.Errorf("log storage has been disposed") } + if ls.watcher == nil { + watchErr := ls.watchResourceEvents(log) + if watchErr != nil { + return nil, watchErr + } + } + resourceWatchers, _ := ls.resourceWatchers.LoadOrStoreNew(uid, func() *syncmap.Map[*resourceWatcher, bool] { return &syncmap.Map[*resourceWatcher, bool]{} }) @@ -201,16 +209,8 @@ func (ls *LogStorage) watchResource(uid types.UID, log logr.Logger) (*resourceWa rw.OnStop(func() { resourceWatchers.Delete(rw) }) - - if ls.watcher == nil { - watchErr := ls.watchResourceEvents(log) - if watchErr != nil { - return nil, watchErr - } - } else { - if event, found := ls.mostRecentResourceEvents.Load(uid); found { - rw.Queue(event) - } + if event, found := ls.mostRecentResourceEvents.Load(uid); found { + rw.Queue(event) } _, _ = resourceWatchers.LoadOrStore(rw, true) @@ -266,11 +266,7 @@ func (ls *LogStorage) watchResourceEvents(log logr.Logger) error { return } - // This has to be initiated before making a call to create a new watcher. - // If not, we can end up in a deadlock situation where the storage is trying to deliver - // a watch event (holding the watch set lock in shared mode) while at the same time - // we are making a call to create a new watcher (which tries to take the watch set lock - // in exclusive mode). + // Stop the old watcher before registering its replacement so generations do not overlap. log.V(1).Info("Stopping old parent resource watcher...") stopWatcher(ls.watcher) log.V(1).Info("Old parent resource watcher stopped") @@ -542,15 +538,7 @@ func (ls *LogStorage) NewGetOptions() (runtime.Object, bool, string) { } func stopWatcher(w watch.Interface) { - // We need to drain the watcher result channel because it is an unbuffered channel - // that the Tilt API server storage layer synchronously writes to it as part of handling object changes. - // If we simply abandon the channel with some pending writes, the corresponding update will block forever. - // Even calling Stop() on the watcher will NOT help, because the Stop() method tries to take an exclusive lock - // on the "watch set" that the watcher belongs to before closing the watcher channel, - // and lock is held (in shared mode) by the goroutine that is writing to the channel - // to broadcast object update events. - // By calling drain() on a separate goroutine we ensure that the channel - // will continue to be drained until the call to Stop() succeeds. + // Keep draining until the result channel closes so pending deliveries are not abandoned during shutdown. go drain(w.ResultChan()) w.Stop() diff --git a/api/v1/log_storage_test.go b/api/v1/log_storage_test.go new file mode 100644 index 00000000..8ef6b405 --- /dev/null +++ b/api/v1/log_storage_test.go @@ -0,0 +1,104 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package v1 + +import ( + "context" + "errors" + "io" + "testing" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/require" + apiserver_resource "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource" + metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion" + "k8s.io/apimachinery/pkg/watch" + registry_rest "k8s.io/apiserver/pkg/registry/rest" + + "github.com/microsoft/dcp/pkg/testutil" +) + +// Verifies that LogStorage.watchResource does not allocate a resource watcher when parent +// watch setup fails and can establish the watch on a subsequent request. +func TestLogStorageWatchSetupFailureCanBeRetried(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, 0) + defer cancel() + setupErr := errors.New("test watch setup failure") + parentWatcher := watch.NewRaceFreeFake() + parentStorage := &logStorageWatchSource{watcher: parentWatcher, watchErr: setupErr} + storage, storageErr := NewLogStorage(parentStorage, logStorageTestStreamer{}) + require.NoError(t, storageErr) + t.Cleanup(storage.Destroy) + + resourceWatcher, watchErr := storage.watchResource("resource", logr.Discard()) + require.ErrorIs(t, watchErr, setupErr) + require.Nil(t, resourceWatcher) + require.Nil(t, storage.watcher) + require.True(t, storage.resourceWatchers.Empty()) + + parentStorage.watchErr = nil + retryWatcher, retryErr := storage.watchResource("resource", logr.Discard()) + require.NoError(t, retryErr) + retryWatcher.Stop() + select { + case _, open := <-retryWatcher.ResultChan(): + require.False(t, open) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + + storage.Destroy() + require.True(t, parentWatcher.IsStopped()) +} + +// Verifies that LogStorage.Destroy closes its disposal signal without a parent watcher, +// remains idempotent, and rejects subsequent resource-watch requests. +func TestLogStorageDestroyWithoutParentWatcher(t *testing.T) { + storage, storageErr := NewLogStorage(&logStorageWatchSource{}, logStorageTestStreamer{}) + require.NoError(t, storageErr) + t.Cleanup(storage.Destroy) + + storage.Destroy() + storage.Destroy() + select { + case <-storage.disposeCh: + default: + t.Fatal("log storage disposal signal was not closed") + } + resourceWatcher, watchErr := storage.watchResource("resource", logr.Discard()) + require.ErrorContains(t, watchErr, "disposed") + require.Nil(t, resourceWatcher) + require.True(t, storage.resourceWatchers.Empty()) +} + +type logStorageWatchSource struct { + registry_rest.StandardStorage + watcher watch.Interface + watchErr error +} + +func (source *logStorageWatchSource) Watch(context.Context, *metainternalversion.ListOptions) (watch.Interface, error) { + if source.watchErr != nil { + return nil, source.watchErr + } + return source.watcher, nil +} + +type logStorageTestStreamer struct{} + +func (logStorageTestStreamer) Preflight(apiserver_resource.Object, *LogOptions) error { + return nil +} + +func (logStorageTestStreamer) StreamLogs(context.Context, io.Writer, apiserver_resource.Object, *LogOptions, logr.Logger) (ResourceStreamStatus, <-chan struct{}, error) { + return ResourceStreamStatusDone, nil, nil +} + +func (logStorageTestStreamer) OnResourceUpdated(ResourceWatcherEvent, logr.Logger) {} + +func (logStorageTestStreamer) Dispose() error { + return nil +} diff --git a/go.mod b/go.mod index 07d3a844..6eef6a57 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/spf13/cobra v1.10.2 github.com/spf13/pflag v1.0.10 github.com/stretchr/testify v1.12.1 - github.com/tilt-dev/tilt-apiserver v0.19.1 + github.com/tilt-dev/tilt-apiserver v0.19.2 github.com/tklauser/go-sysconf v0.4.0 go.opentelemetry.io/otel v1.46.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.46.0 diff --git a/go.sum b/go.sum index 73fb40ff..d801ed5f 100644 --- a/go.sum +++ b/go.sum @@ -573,8 +573,8 @@ github.com/tenntenn/text/transform v0.0.0-20200319021203-7eef512accb3 h1:f+jULpR github.com/tenntenn/text/transform v0.0.0-20200319021203-7eef512accb3/go.mod h1:ON8b8w4BN/kE1EOhwT0o+d62W65a6aPw1nouo9LMgyY= github.com/tetafro/godot v1.5.4 h1:u1ww+gqpRLiIA16yF2PV1CV1n/X3zhyezbNXC3E14Sg= github.com/tetafro/godot v1.5.4/go.mod h1:eOkMrVQurDui411nBY2FA05EYH01r14LuWY/NrVDVcU= -github.com/tilt-dev/tilt-apiserver v0.19.1 h1:j6B/EpzgbJuMCLo/lvskPowHx5oFgQT7VXTyzTHe7zw= -github.com/tilt-dev/tilt-apiserver v0.19.1/go.mod h1:w6vnH5H5npxMvy+iVj+DBqtiFV9TooSOZh/7shv4Njo= +github.com/tilt-dev/tilt-apiserver v0.19.2 h1:0bujknyw1XYO648mzFbCeG64pwHVLYszHYjIsRiRLas= +github.com/tilt-dev/tilt-apiserver v0.19.2/go.mod h1:w6vnH5H5npxMvy+iVj+DBqtiFV9TooSOZh/7shv4Njo= github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67 h1:9LPGD+jzxMlnk5r6+hJnar67cgpDIz/iyD+rfl5r2Vk= github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67/go.mod h1:mkjARE7Yr8qU23YcGMSALbIxTQ9r9QBVahQOBRfU460= github.com/timonwong/loggercheck v0.11.0 h1:jdaMpYBl+Uq9mWPXv1r8jc5fC3gyXx4/WGwTnnNKn4M= diff --git a/internal/apiserver/v2_namespace_lifecycle_storage_test.go b/internal/apiserver/v2_namespace_lifecycle_storage_test.go index 47cb95ca..4c2e7244 100644 --- a/internal/apiserver/v2_namespace_lifecycle_storage_test.go +++ b/internal/apiserver/v2_namespace_lifecycle_storage_test.go @@ -8,7 +8,6 @@ package apiserver import ( "context" "errors" - "sync" "testing" "github.com/stretchr/testify/require" @@ -81,51 +80,6 @@ func TestV2NamespaceLifecycleStorageHoldsChildLeaseAfterHTTPTimeout(t *testing.T requireNoV2NamespaceMutation(t, fixture.gate, "test") } -func TestV2NamespaceLifecycleStorageHoldsDeleteLeaseAfterHTTPTimeout(t *testing.T) { - fixture := newV2NamespaceStorageFixture(t) - fixture.createNamespace(t) - namespaceWatch, watchErr := fixture.namespaces.Watch(fixture.ctx, nil) - require.NoError(t, watchErr) - stopWatch := sync.OnceFunc(func() { stopV2NamespaceWatcher(namespaceWatch) }) - defer stopWatch() - persisted := make(chan struct{}) - fixture.fs.afterWrite = func(obj runtime.Object) { - if namespace, namespaceOK := obj.(*apiv2.Namespace); namespaceOK && namespace.DeletionTimestamp != nil { - close(persisted) - } - } - requestCtx, cancelRequest := context.WithCancel(fixture.ctx) - defer cancelRequest() - httpResult := make(chan error, 1) - storageDone := make(chan struct{}) - go func() { - _, finishErr := finisher.FinishRequest(requestCtx, func() (runtime.Object, error) { - defer close(storageDone) - deleted, _, deleteErr := fixture.namespaces.Delete(requestCtx, "test", nil, nil) - return deleted, deleteErr - }) - httpResult <- finishErr - }() - waitForSignal(t, fixture.ctx, persisted) - cancelRequest() - require.True(t, apierrors.IsTimeout(waitForError(t, fixture.ctx, httpResult))) - fixture.requireTerminating(t) - waitV2NamespaceMutationReferences(t, fixture.ctx, fixture.gate, "test", 1) - require.Empty(t, fixture.gate.closedNamespaceStates()) - _, createErr := fixture.children.Create(fixture.ctx, newV2LifecycleChild(), nil, nil) - require.True(t, apierrors.IsForbidden(createErr)) - - stopWatch() - waitForSignal(t, fixture.ctx, storageDone) - requireNoV2NamespaceMutation(t, fixture.gate, "test") - fixture.gate.lock.Lock() - defer fixture.gate.lock.Unlock() - state := fixture.gate.namespaces["test"] - require.True(t, state.closed) - require.True(t, state.deleteAccepted) - require.Zero(t, state.activeDeletes) -} - func TestV2NamespaceLifecycleStorageSerializesNamespaceCreateAndDelete(t *testing.T) { fixture := newV2NamespaceStorageFixture(t) writeStarted, releaseWrite := make(chan struct{}), make(chan struct{}) diff --git a/internal/apiserver/v2_namespace_lifecycle_watcher.go b/internal/apiserver/v2_namespace_lifecycle_watcher.go index e7cd941c..b99b325c 100644 --- a/internal/apiserver/v2_namespace_lifecycle_watcher.go +++ b/internal/apiserver/v2_namespace_lifecycle_watcher.go @@ -67,10 +67,15 @@ func runV2NamespaceLifecycleWatcher( continue } + deletions := make(chan struct{}, 1) + watchErrors := make(chan struct{}, 1) + watchDone := make(chan struct{}) + go readV2NamespaceWatchEvents(namespaceWatcher, deletions, watchErrors, watchDone, log) + closedStates := gate.closedNamespaceStates() namespaceList, listErr := source.List(ctx, metav1.ListOptions{}) if listErr != nil { - stopV2NamespaceWatcher(namespaceWatcher) + stopV2NamespaceWatch(namespaceWatcher, watchDone) if ctx.Err() != nil { return } @@ -82,13 +87,12 @@ func runV2NamespaceLifecycleWatcher( restartTimer := time.NewTimer(restartInterval) restartDelay = time.Duration(0) - watcherClosed := false watchLoop: for { select { case <-ctx.Done(): restartTimer.Stop() - stopV2NamespaceWatcher(namespaceWatcher) + stopV2NamespaceWatch(namespaceWatcher, watchDone) return case <-restartTimer.C: log.V(1).Info("Restarting V2 Namespace lifecycle watcher") @@ -96,21 +100,14 @@ func runV2NamespaceLifecycleWatcher( case <-gate.refreshRequested: log.V(1).Info("Refreshing V2 Namespace lifecycle state") break watchLoop - case event, open := <-namespaceWatcher.ResultChan(): - if !open { - log.V(1).Info("V2 Namespace lifecycle watch stream closed") - restartDelay = retryInterval - watcherClosed = true - break watchLoop - } - if event.Type == watch.Error { - log.Info("V2 Namespace lifecycle watcher received an error event", "Object", event.Object) - restartDelay = retryInterval - break watchLoop - } - if event.Type != watch.Deleted { - continue - } + case <-watchDone: + log.V(1).Info("V2 Namespace lifecycle watch stream closed") + restartDelay = retryInterval + break watchLoop + case <-watchErrors: + restartDelay = retryInterval + break watchLoop + case <-deletions: log.V(1).Info("Observed V2 Namespace storage deletion") break watchLoop } @@ -121,12 +118,46 @@ func runV2NamespaceLifecycleWatcher( default: } } - if !watcherClosed { - stopV2NamespaceWatcher(namespaceWatcher) + stopV2NamespaceWatch(namespaceWatcher, watchDone) + } +} + +func readV2NamespaceWatchEvents( + namespaceWatcher watch.Interface, + deletions, watchErrors, watchDone chan<- struct{}, + log logr.Logger, +) { + defer close(watchDone) + for event := range namespaceWatcher.ResultChan() { + switch event.Type { + case watch.Deleted: + // Deletions only request a fresh snapshot, so one pending notification is enough. + select { + case deletions <- struct{}{}: + default: + } + case watch.Error: + log.Info("V2 Namespace lifecycle watcher received an error event", "Object", event.Object) + select { + case watchErrors <- struct{}{}: + default: + } } } } +// stopV2NamespaceWatch stops a watch with an active reader and waits for that reader to exit. +func stopV2NamespaceWatch(namespaceWatcher watch.Interface, watchDone <-chan struct{}) { + select { + case <-watchDone: + return + default: + } + // Stop may return before the result channel closes; join its reader before restarting. + namespaceWatcher.Stop() + <-watchDone +} + func v2NamespaceStates(namespaceList *unstructured.UnstructuredList) map[string]v2NamespaceStorageState { namespaces := make(map[string]v2NamespaceStorageState, len(namespaceList.Items)) for index := range namespaceList.Items { @@ -140,7 +171,7 @@ func v2NamespaceStates(namespaceList *unstructured.UnstructuredList) map[string] func waitForV2NamespaceWatcherRestart(ctx context.Context, delay time.Duration) bool { if delay == 0 { - return true + return ctx.Err() == nil } timer := time.NewTimer(delay) defer timer.Stop() @@ -152,15 +183,3 @@ func waitForV2NamespaceWatcherRestart(ctx context.Context, delay time.Duration) return true } } - -func stopV2NamespaceWatcher(namespaceWatcher watch.Interface) { - // Tilt broadcasts on an unbuffered channel while holding the WatchSet lock that Stop needs. - // Keep draining until Stop closes the channel so an in-progress broadcast cannot deadlock. - go drainV2NamespaceWatcher(namespaceWatcher.ResultChan()) - namespaceWatcher.Stop() -} - -func drainV2NamespaceWatcher(events <-chan watch.Event) { - for range events { - } -} diff --git a/internal/apiserver/v2_namespace_lifecycle_watcher_test.go b/internal/apiserver/v2_namespace_lifecycle_watcher_test.go index ab6ad633..c098ae33 100644 --- a/internal/apiserver/v2_namespace_lifecycle_watcher_test.go +++ b/internal/apiserver/v2_namespace_lifecycle_watcher_test.go @@ -7,6 +7,7 @@ package apiserver import ( "context" + "errors" "sync" "testing" "time" @@ -218,13 +219,200 @@ func TestV2NamespaceLifecycleWatcherObservesDeletionCompletedAfterList(t *testin waitForV2NamespaceGateRemoval(t, ctx, gate, "test") } +// Verifies that runV2NamespaceLifecycleWatcher drains unbuffered events while a list is blocked +// and refreshes lifecycle state after coalesced deletions. +func TestV2NamespaceLifecycleWatcherDrainsWhileListing(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, v2NamespaceLifecycleTestTimeout) + t.Cleanup(cancel) + watcherCtx, cancelWatcher := context.WithCancel(ctx) + + gate := closedV2NamespaceLifecycleGate(t, ctx, "test") + source := newFakeV2NamespaceWatchSource("test") + events := make(chan watch.Event) + stopped := make(chan struct{}) + source.firstWatcher = watch.MockWatcher{ + StopFunc: sync.OnceFunc(func() { + close(events) + close(stopped) + }), + ResultChanFunc: func() <-chan watch.Event { return events }, + } + listCompleted := make(chan struct{}) + returnList := make(chan struct{}) + releaseList := sync.OnceFunc(func() { close(returnList) }) + var blockFirstList sync.Once + source.afterList = func() { + blockFirstList.Do(func() { + close(listCompleted) + select { + case <-returnList: + case <-watcherCtx.Done(): + } + }) + } + + watcherDone := make(chan struct{}) + go func() { + defer close(watcherDone) + runV2NamespaceLifecycleWatcher(watcherCtx, source, gate, logr.Discard(), time.Hour, time.Millisecond) + }() + t.Cleanup(func() { + cancelWatcher() + waitForV2NamespaceWatcherSignal(t, ctx, watcherDone) + }) + t.Cleanup(releaseList) + waitForV2NamespaceWatcherSignal(t, ctx, listCompleted) + + source.setNamespaces() + for _, eventType := range []watch.EventType{watch.Added, watch.Modified, watch.Deleted, watch.Deleted, watch.Modified} { + sendV2NamespaceWatcherEvent(t, ctx, events, watch.Event{Type: eventType, Object: v2NamespaceObject("test")}) + } + requireV2NamespaceGateState(t, gate, "test") + releaseList() + + waitForV2NamespaceWatcherSignal(t, ctx, stopped) + waitForV2NamespaceGateRemoval(t, ctx, gate, "test") +} + +// Verifies that runV2NamespaceLifecycleWatcher preserves watch errors and closure during a +// blocked list and continues draining after an error until the watch is stopped. +func TestV2NamespaceLifecycleWatcherRefreshesAfterWatchFailureDuringList(t *testing.T) { + for _, failure := range []string{"error", "closed"} { + t.Run(failure, func(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, v2NamespaceLifecycleTestTimeout) + t.Cleanup(cancel) + watcherCtx, cancelWatcher := context.WithCancel(ctx) + gate := closedV2NamespaceLifecycleGate(t, ctx, "test") + source := newFakeV2NamespaceWatchSource("test") + events := make(chan watch.Event) + stopEvents := sync.OnceFunc(func() { close(events) }) + source.firstWatcher = watch.MockWatcher{ + StopFunc: stopEvents, + ResultChanFunc: func() <-chan watch.Event { return events }, + } + listCompleted := make(chan struct{}) + returnList := make(chan struct{}) + releaseList := sync.OnceFunc(func() { close(returnList) }) + var blockFirstList sync.Once + source.afterList = func() { + blockFirstList.Do(func() { + close(listCompleted) + select { + case <-returnList: + case <-watcherCtx.Done(): + } + }) + } + watcherDone := make(chan struct{}) + go func() { + defer close(watcherDone) + runV2NamespaceLifecycleWatcher(watcherCtx, source, gate, logr.Discard(), time.Hour, time.Millisecond) + }() + t.Cleanup(func() { + cancelWatcher() + waitForV2NamespaceWatcherSignal(t, ctx, watcherDone) + }) + t.Cleanup(releaseList) + waitForV2NamespaceWatcherSignal(t, ctx, listCompleted) + + source.setNamespaces() + if failure == "error" { + sendV2NamespaceWatcherEvent(t, ctx, events, watch.Event{ + Type: watch.Error, Object: &metav1.Status{Message: "test watch failure"}, + }) + sendV2NamespaceWatcherEvent(t, ctx, events, watch.Event{Type: watch.Modified, Object: v2NamespaceObject("test")}) + } else { + stopEvents() + } + releaseList() + + waitForFakeV2NamespaceWatcher(t, ctx, source.watchers) + waitForV2NamespaceGateRemoval(t, ctx, gate, "test") + }) + } +} + +// Verifies that runV2NamespaceLifecycleWatcher stops its watch and reader when the initial +// list fails or its lifetime context is canceled during the list. +func TestV2NamespaceLifecycleWatcherStopsWatchAfterListFailureOrCancellation(t *testing.T) { + for _, failure := range []string{"list-error", "canceled"} { + t.Run(failure, func(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, v2NamespaceLifecycleTestTimeout) + t.Cleanup(cancel) + watcherCtx, cancelWatcher := context.WithCancel(ctx) + source := newFakeV2NamespaceWatchSource("test") + source.firstListErr = errors.New("test list failure") + events := make(chan watch.Event) + stopped := make(chan struct{}) + source.firstWatcher = watch.MockWatcher{ + StopFunc: sync.OnceFunc(func() { + close(events) + close(stopped) + }), + ResultChanFunc: func() <-chan watch.Event { return events }, + } + listStarted := make(chan struct{}) + continueList := make(chan struct{}) + releaseList := sync.OnceFunc(func() { close(continueList) }) + var blockFirstList sync.Once + source.beforeList = func() { + blockFirstList.Do(func() { + close(listStarted) + select { + case <-continueList: + case <-watcherCtx.Done(): + } + }) + } + watcherDone := make(chan struct{}) + go func() { + defer close(watcherDone) + runV2NamespaceLifecycleWatcher(watcherCtx, source, newV2NamespaceLifecycleGate(), logr.Discard(), time.Hour, time.Millisecond) + }() + t.Cleanup(func() { + cancelWatcher() + waitForV2NamespaceWatcherSignal(t, ctx, watcherDone) + }) + t.Cleanup(releaseList) + waitForV2NamespaceWatcherSignal(t, ctx, listStarted) + sendV2NamespaceWatcherEvent(t, ctx, events, watch.Event{Type: watch.Modified, Object: v2NamespaceObject("test")}) + + if failure == "canceled" { + cancelWatcher() + } else { + releaseList() + } + waitForV2NamespaceWatcherSignal(t, ctx, stopped) + cancelWatcher() + waitForV2NamespaceWatcherSignal(t, ctx, watcherDone) + }) + } +} + +// Verifies that runV2NamespaceLifecycleWatcher does not open a watch after cancellation. +func TestV2NamespaceLifecycleWatcherSkipsWatchAfterCancellation(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, v2NamespaceLifecycleTestTimeout) + cancel() + source := newFakeV2NamespaceWatchSource() + + runV2NamespaceLifecycleWatcher(ctx, source, newV2NamespaceLifecycleGate(), logr.Discard(), time.Hour, time.Millisecond) + + select { + case <-source.watchers: + t.Fatal("opened a watch after cancellation") + default: + } +} + type fakeV2NamespaceWatchSource struct { - lock sync.Mutex - namespaces map[string]bool - watchers chan *watch.RaceFreeFakeWatcher - lists chan struct{} - beforeList func() - afterList func() + lock sync.Mutex + namespaces map[string]bool + watchers chan *watch.RaceFreeFakeWatcher + lists chan struct{} + beforeList func() + afterList func() + firstWatcher watch.Interface + firstListErr error } func newFakeV2NamespaceWatchSource(namespaces ...string) *fakeV2NamespaceWatchSource { @@ -238,14 +426,23 @@ func newFakeV2NamespaceWatchSource(namespaces ...string) *fakeV2NamespaceWatchSo } func (source *fakeV2NamespaceWatchSource) List( - _ context.Context, + ctx context.Context, _ metav1.ListOptions, ) (*unstructured.UnstructuredList, error) { if source.beforeList != nil { source.beforeList() } + if ctx.Err() != nil { + return nil, ctx.Err() + } source.lock.Lock() + if source.firstListErr != nil { + listErr := source.firstListErr + source.firstListErr = nil + source.lock.Unlock() + return nil, listErr + } namespaceList := &unstructured.UnstructuredList{} for namespace, terminating := range source.namespaces { namespaceObject := v2NamespaceObject(namespace) @@ -268,6 +465,13 @@ func (source *fakeV2NamespaceWatchSource) Watch( _ context.Context, _ metav1.ListOptions, ) (watch.Interface, error) { + source.lock.Lock() + firstWatcher := source.firstWatcher + source.firstWatcher = nil + source.lock.Unlock() + if firstWatcher != nil { + return firstWatcher, nil + } namespaceWatcher := watch.NewRaceFreeFake() source.watchers <- namespaceWatcher return namespaceWatcher, nil @@ -370,3 +574,13 @@ func waitForV2NamespaceWatcherSignal( t.Fatal(ctx.Err()) } } + +func sendV2NamespaceWatcherEvent(t *testing.T, ctx context.Context, events chan<- watch.Event, event watch.Event) { + t.Helper() + + select { + case events <- event: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } +}