[release/0.26] Harden watcher consumption and adopt bounded Tilt watch delivery

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

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()

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
+}

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

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=

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{})