From 94dd61ec7c8e6bd2fa560a4071526ace300ef98e Mon Sep 17 00:00:00 2001 From: David Negstad Date: Wed, 23 Sep 2026 19:19:58 -0700 Subject: [PATCH] Serialize API resource mutations Prevent concurrent mutations of the same API resource type from publishing watch events out of storage commit order. Add deterministic coverage that reproduces the Tilt filepath storage race and verifies the middleware preserves ordering. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- internal/apiserver/apiserver.go | 1 + .../ordered_resource_mutation_handler.go | 63 +++ .../ordered_resource_mutation_handler_test.go | 172 ++++++++ .../ordered_resource_mutation_storage_test.go | 383 ++++++++++++++++++ 4 files changed, 619 insertions(+) create mode 100644 internal/apiserver/ordered_resource_mutation_handler.go create mode 100644 internal/apiserver/ordered_resource_mutation_handler_test.go create mode 100644 internal/apiserver/ordered_resource_mutation_storage_test.go diff --git a/internal/apiserver/apiserver.go b/internal/apiserver/apiserver.go index a10f6ec5..a0fcd1b6 100644 --- a/internal/apiserver/apiserver.go +++ b/internal/apiserver/apiserver.go @@ -380,6 +380,7 @@ func addDcpHttpHandlers( return namespaceWatchSourceErr } config.GenericConfig.BuildHandlerChainFunc = func(handler http.Handler, c *kubeapiserver.Config) http.Handler { + handler = withOrderedResourceMutations(handler) handler = originalChainBuilder(handler, c) handler = withDcpContextValues(handler, ctx, log) return handler diff --git a/internal/apiserver/ordered_resource_mutation_handler.go b/internal/apiserver/ordered_resource_mutation_handler.go new file mode 100644 index 00000000..5a43ab67 --- /dev/null +++ b/internal/apiserver/ordered_resource_mutation_handler.go @@ -0,0 +1,63 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package apiserver + +import ( + "net/http" + "sync" + + genericapirequest "k8s.io/apiserver/pkg/endpoints/request" + + "github.com/microsoft/dcp/pkg/syncmap" +) + +type resourceMutationKey struct { + apiGroup string + resource string +} + +type orderedResourceMutationHandler struct { + inner http.Handler + locks *syncmap.Map[resourceMutationKey, sync.Locker] +} + +func withOrderedResourceMutations(handler http.Handler) http.Handler { + return &orderedResourceMutationHandler{ + inner: handler, + locks: &syncmap.Map[resourceMutationKey, sync.Locker]{}, + } +} + +func (h *orderedResourceMutationHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + requestInfo, found := genericapirequest.RequestInfoFrom(r.Context()) + if !found || !requestInfo.IsResourceRequest || !isResourceMutation(requestInfo.Verb) { + h.inner.ServeHTTP(w, r) + return + } + + // Tilt's storage shares one watch stream across versions and subresources of a GroupResource. + // Serialize the complete mutation, including watcher notification, on the same boundary. + key := resourceMutationKey{ + apiGroup: requestInfo.APIGroup, + resource: requestInfo.Resource, + } + mutationLock, _ := h.locks.LoadOrStoreNew(key, func() sync.Locker { + return &sync.Mutex{} + }) + + mutationLock.Lock() + defer mutationLock.Unlock() + h.inner.ServeHTTP(w, r) +} + +func isResourceMutation(verb string) bool { + switch verb { + case "create", "update", "patch", "delete", "deletecollection": + return true + default: + return false + } +} diff --git a/internal/apiserver/ordered_resource_mutation_handler_test.go b/internal/apiserver/ordered_resource_mutation_handler_test.go new file mode 100644 index 00000000..9abe6d9a --- /dev/null +++ b/internal/apiserver/ordered_resource_mutation_handler_test.go @@ -0,0 +1,172 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package apiserver + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + genericapirequest "k8s.io/apiserver/pkg/endpoints/request" + + "github.com/microsoft/dcp/pkg/testutil" +) + +type signalingLocker struct { + lockAttempted chan struct{} + release chan struct{} + once sync.Once +} + +func newSignalingLocker() *signalingLocker { + return &signalingLocker{ + lockAttempted: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (l *signalingLocker) Lock() { + l.once.Do(func() { + close(l.lockAttempted) + }) + <-l.release +} + +func (l *signalingLocker) Unlock() {} + +func TestOrderedResourceMutationHandlerSerializesStatusMutationWithResource(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, time.Minute) + defer cancel() + + handled := make(chan struct{}) + handler := withOrderedResourceMutations(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + close(handled) + })).(*orderedResourceMutationHandler) + lock := newSignalingLocker() + handler.locks.Store(resourceMutationKey{ + apiGroup: "usvc-dev.developer.microsoft.com", + resource: "services", + }, lock) + + request := newResourceRequest(ctx, "update", "usvc-dev.developer.microsoft.com", "services", "status") + go handler.ServeHTTP(httptest.NewRecorder(), request) + + select { + case <-lock.lockAttempted: + case <-ctx.Done(): + t.Fatal("timed out waiting for the resource mutation lock") + } + + select { + case <-handled: + t.Fatal("status mutation reached storage before the resource mutation lock was released") + default: + } + + close(lock.release) + select { + case <-handled: + case <-ctx.Done(): + t.Fatal("timed out waiting for the serialized mutation") + } +} + +func TestOrderedResourceMutationHandlerAllowsIndependentRequests(t *testing.T) { + t.Parallel() + + testCases := map[string]struct { + requestInfo *genericapirequest.RequestInfo + }{ + "resource read": { + requestInfo: &genericapirequest.RequestInfo{ + IsResourceRequest: true, + Verb: "get", + APIGroup: "usvc-dev.developer.microsoft.com", + Resource: "services", + }, + }, + "different resource mutation": { + requestInfo: &genericapirequest.RequestInfo{ + IsResourceRequest: true, + Verb: "update", + APIGroup: "usvc-dev.developer.microsoft.com", + Resource: "executables", + }, + }, + "non-resource mutation": { + requestInfo: &genericapirequest.RequestInfo{ + Verb: "patch", + }, + }, + } + + for name, testCase := range testCases { + t.Run(name, func(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, time.Minute) + defer cancel() + + handled := make(chan struct{}) + handler := withOrderedResourceMutations(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + close(handled) + })).(*orderedResourceMutationHandler) + lock := newSignalingLocker() + handler.locks.Store(resourceMutationKey{ + apiGroup: "usvc-dev.developer.microsoft.com", + resource: "services", + }, lock) + + requestCtx := genericapirequest.WithRequestInfo(ctx, testCase.requestInfo) + request := httptest.NewRequestWithContext(requestCtx, http.MethodGet, "/", nil) + go handler.ServeHTTP(httptest.NewRecorder(), request) + + select { + case <-handled: + case <-lock.lockAttempted: + close(lock.release) + t.Fatal("independent request acquired the Service mutation lock") + case <-ctx.Done(): + t.Fatal("timed out waiting for the independent request") + } + }) + } +} + +func TestIsResourceMutation(t *testing.T) { + t.Parallel() + + for _, verb := range []string{"create", "update", "patch", "delete", "deletecollection"} { + require.True(t, isResourceMutation(verb), "expected %q to be treated as a mutation", verb) + } + for _, verb := range []string{"get", "list", "watch", "connect", ""} { + require.False(t, isResourceMutation(verb), "expected %q not to be treated as a mutation", verb) + } +} + +func newResourceRequest( + ctx context.Context, + verb string, + apiGroup string, + resource string, + subresource string, +) *http.Request { + requestInfo := &genericapirequest.RequestInfo{ + IsResourceRequest: true, + Verb: verb, + APIGroup: apiGroup, + Resource: resource, + Subresource: subresource, + } + requestCtx := genericapirequest.WithRequestInfo(ctx, requestInfo) + return httptest.NewRequestWithContext(requestCtx, http.MethodPatch, "/", nil) +} diff --git a/internal/apiserver/ordered_resource_mutation_storage_test.go b/internal/apiserver/ordered_resource_mutation_storage_test.go new file mode 100644 index 00000000..f1431c03 --- /dev/null +++ b/internal/apiserver/ordered_resource_mutation_storage_test.go @@ -0,0 +1,383 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package apiserver + +import ( + "context" + "net/http" + "net/http/httptest" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + tiltresource "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource" + tiltrest "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/rest" + tiltfilepath "github.com/tilt-dev/tilt-apiserver/pkg/storage/filepath" + metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/apimachinery/pkg/watch" + genericapirequest "k8s.io/apiserver/pkg/endpoints/request" + "k8s.io/apiserver/pkg/registry/rest" + + apiv1 "github.com/microsoft/dcp/api/v1" + "github.com/microsoft/dcp/pkg/testutil" +) + +const mutationOrderingTestNamespace = "test" + +type mutationOrderingFS struct { + tiltfilepath.FS + + blockStatusWrite atomic.Bool + statusWriteCommitted chan struct{} + deletionWriteCommitted chan struct{} + releaseStatusWrite chan struct{} + statusCommitOnce sync.Once + deletionCommitOnce sync.Once + releaseOnce sync.Once +} + +func newMutationOrderingFS() *mutationOrderingFS { + fs := &mutationOrderingFS{ + FS: tiltfilepath.NewMemoryFS(), + statusWriteCommitted: make(chan struct{}), + deletionWriteCommitted: make(chan struct{}), + releaseStatusWrite: make(chan struct{}), + } + fs.blockStatusWrite.Store(true) + return fs +} + +func (fs *mutationOrderingFS) Write( + encoder runtime.Encoder, + path string, + obj runtime.Object, + storageVersion uint64, +) error { + writeErr := fs.FS.Write(encoder, path, obj, storageVersion) + if writeErr != nil { + return writeErr + } + + service, isService := obj.(*apiv1.Service) + if !isService { + return nil + } + if service.DeletionTimestamp != nil && !service.DeletionTimestamp.IsZero() { + fs.deletionCommitOnce.Do(func() { + close(fs.deletionWriteCommitted) + }) + return nil + } + if service.Status.State == apiv1.ServiceStateReady && fs.blockStatusWrite.CompareAndSwap(true, false) { + fs.statusCommitOnce.Do(func() { + close(fs.statusWriteCommitted) + }) + <-fs.releaseStatusWrite + } + return nil +} + +func (fs *mutationOrderingFS) releaseStatus() { + fs.releaseOnce.Do(func() { + close(fs.releaseStatusWrite) + }) +} + +type mutationOrderingFixture struct { + ctx context.Context + service *apiv1.Service + parentStorage rest.Storage + statusStorage rest.Storage + resourceWatch watch.Interface + watchEvents chan watch.Event + fs *mutationOrderingFS + statusResult chan error + deletionResult chan error +} + +func newMutationOrderingFixture(t *testing.T, ctx context.Context) *mutationOrderingFixture { + t.Helper() + + scheme := runtime.NewScheme() + require.NoError(t, apiv1.AddToScheme(scheme)) + metav1.AddToGroupVersion(scheme, apiv1.GroupVersion) + codec := serializer.NewCodecFactory(scheme).LegacyCodec(apiv1.GroupVersion) + fs := newMutationOrderingFS() + watchSet := tiltfilepath.NewWatchSet() + serviceResource := &apiv1.Service{} + defaultStrategy := tiltrest.DefaultStrategy{ + Object: serviceResource, + ObjectTyper: scheme, + } + parentStorage := tiltfilepath.NewFilepathREST( + fs, + watchSet, + defaultStrategy, + serviceResource.GetGroupVersionResource().GroupResource(), + codec, + "data", + serviceResource.New, + serviceResource.NewList, + ) + statusStorage := tiltfilepath.NewFilepathREST( + fs, + watchSet, + tiltrest.StatusSubResourceStrategy{Strategy: defaultStrategy}, + serviceResource.GetGroupVersionResource().GroupResource(), + codec, + "data", + serviceResource.New, + serviceResource.NewList, + ) + namespacedCtx := genericapirequest.WithNamespace(ctx, mutationOrderingTestNamespace) + service := &apiv1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "service", + Namespace: mutationOrderingTestNamespace, + Finalizers: []string{"test.dcp.microsoft.com/finalizer"}, + }, + } + created, createErr := parentStorage.(rest.Creater).Create( + namespacedCtx, + service, + nil, + &metav1.CreateOptions{}, + ) + require.NoError(t, createErr) + service = created.(*apiv1.Service) + + resourceWatch, watchErr := parentStorage.(rest.Watcher).Watch( + namespacedCtx, + &metainternalversion.ListOptions{}, + ) + require.NoError(t, watchErr) + watchEvents := make(chan watch.Event, 3) + go func() { + for event := range resourceWatch.ResultChan() { + watchEvents <- event + } + }() + + fixture := &mutationOrderingFixture{ + ctx: namespacedCtx, + service: service, + parentStorage: parentStorage, + statusStorage: statusStorage, + resourceWatch: resourceWatch, + watchEvents: watchEvents, + fs: fs, + statusResult: make(chan error, 1), + deletionResult: make(chan error, 1), + } + t.Cleanup(func() { + fs.releaseStatus() + resourceWatch.Stop() + }) + + initialEvent := fixture.nextWatchEvent(t) + require.Equal(t, watch.Added, initialEvent.Type) + return fixture +} + +func (f *mutationOrderingFixture) handler() http.Handler { + return http.HandlerFunc(func(_ http.ResponseWriter, request *http.Request) { + switch request.Header.Get("test-operation") { + case "status": + updated := f.service.DeepCopy() + updated.Status.State = apiv1.ServiceStateReady + _, _, updateErr := f.statusStorage.(rest.Updater).Update( + f.ctx, + updated.Name, + rest.DefaultUpdatedObjectInfo(updated), + nil, + nil, + false, + &metav1.UpdateOptions{}, + ) + f.statusResult <- updateErr + case "delete": + _, _, deleteErr := f.parentStorage.(rest.GracefulDeleter).Delete( + f.ctx, + f.service.Name, + nil, + &metav1.DeleteOptions{}, + ) + f.deletionResult <- deleteErr + } + }) +} + +func (f *mutationOrderingFixture) request(operation string, verb string, subresource string) *http.Request { + requestInfo := &genericapirequest.RequestInfo{ + IsResourceRequest: true, + Verb: verb, + APIGroup: apiv1.GroupVersion.Group, + APIVersion: apiv1.GroupVersion.Version, + Namespace: mutationOrderingTestNamespace, + Resource: f.service.GetGroupVersionResource().Resource, + Subresource: subresource, + Name: f.service.Name, + } + requestCtx := genericapirequest.WithRequestInfo(f.ctx, requestInfo) + request := httptest.NewRequestWithContext(requestCtx, http.MethodPatch, "/", nil) + request.Header.Set("test-operation", operation) + return request +} + +func (f *mutationOrderingFixture) nextWatchEvent(t *testing.T) watch.Event { + t.Helper() + select { + case event := <-f.watchEvents: + return event + case <-f.ctx.Done(): + t.Fatal("timed out waiting for watch event") + return watch.Event{} + } +} + +func (f *mutationOrderingFixture) requireMutationResults(t *testing.T) { + t.Helper() + select { + case statusErr := <-f.statusResult: + require.NoError(t, statusErr) + case <-f.ctx.Done(): + t.Fatal("timed out waiting for status update") + } + select { + case deletionErr := <-f.deletionResult: + require.NoError(t, deletionErr) + case <-f.ctx.Done(): + t.Fatal("timed out waiting for deletion") + } +} + +func (f *mutationOrderingFixture) requireStoredServiceDeleting(t *testing.T) { + t.Helper() + stored, getErr := f.parentStorage.(rest.Getter).Get(f.ctx, f.service.Name, &metav1.GetOptions{}) + require.NoError(t, getErr) + require.NotNil(t, stored.(*apiv1.Service).DeletionTimestamp) +} + +type observingMutex struct { + sync.Mutex + attempts atomic.Int32 + secondAttempt chan struct{} + secondOnce sync.Once +} + +func newObservingMutex() *observingMutex { + return &observingMutex{secondAttempt: make(chan struct{})} +} + +func (m *observingMutex) Lock() { + if m.attempts.Add(1) == 2 { + m.secondOnce.Do(func() { + close(m.secondAttempt) + }) + } + m.Mutex.Lock() +} + +func TestOrderedResourceMutationHandlerPreventsTiltWatchEventReordering(t *testing.T) { + t.Parallel() + + t.Run("unserialized Tilt storage publishes stale event after deletion", func(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, time.Minute) + defer cancel() + fixture := newMutationOrderingFixture(t, ctx) + handler := fixture.handler() + + go handler.ServeHTTP(httptest.NewRecorder(), fixture.request("status", "update", "status")) + select { + case <-fixture.fs.statusWriteCommitted: + case <-ctx.Done(): + t.Fatal("timed out waiting for status write") + } + + go handler.ServeHTTP(httptest.NewRecorder(), fixture.request("delete", "delete", "")) + select { + case <-fixture.fs.deletionWriteCommitted: + case <-ctx.Done(): + t.Fatal("timed out waiting for deletion write") + } + + deletionEvent := fixture.nextWatchEvent(t) + fixture.fs.releaseStatus() + staleStatusEvent := fixture.nextWatchEvent(t) + fixture.requireMutationResults(t) + fixture.requireStoredServiceDeleting(t) + + require.NotNil(t, deletionEvent.Object.(*apiv1.Service).DeletionTimestamp) + require.Nil(t, staleStatusEvent.Object.(*apiv1.Service).DeletionTimestamp) + require.Greater( + t, + resourceVersion(t, deletionEvent), + resourceVersion(t, staleStatusEvent), + "Tilt published an older non-deleting object after the deletion event", + ) + }) + + t.Run("middleware preserves storage and watch event order", func(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, time.Minute) + defer cancel() + fixture := newMutationOrderingFixture(t, ctx) + handler := withOrderedResourceMutations(fixture.handler()).(*orderedResourceMutationHandler) + mutationLock := newObservingMutex() + handler.locks.Store(resourceMutationKey{ + apiGroup: apiv1.GroupVersion.Group, + resource: fixture.service.GetGroupVersionResource().Resource, + }, mutationLock) + + go handler.ServeHTTP(httptest.NewRecorder(), fixture.request("status", "update", "status")) + select { + case <-fixture.fs.statusWriteCommitted: + case <-ctx.Done(): + t.Fatal("timed out waiting for status write") + } + + go handler.ServeHTTP(httptest.NewRecorder(), fixture.request("delete", "delete", "")) + select { + case <-mutationLock.secondAttempt: + case <-ctx.Done(): + t.Fatal("timed out waiting for deletion to contend on the mutation lock") + } + select { + case <-fixture.fs.deletionWriteCommitted: + t.Fatal("deletion reached storage before the status event was published") + default: + } + + fixture.fs.releaseStatus() + statusEvent := fixture.nextWatchEvent(t) + deletionEvent := fixture.nextWatchEvent(t) + fixture.requireMutationResults(t) + fixture.requireStoredServiceDeleting(t) + + require.Nil(t, statusEvent.Object.(*apiv1.Service).DeletionTimestamp) + require.NotNil(t, deletionEvent.Object.(*apiv1.Service).DeletionTimestamp) + require.Less( + t, + resourceVersion(t, statusEvent), + resourceVersion(t, deletionEvent), + "middleware did not preserve mutation commit order in the watch stream", + ) + }) +} + +func resourceVersion(t *testing.T, event watch.Event) uint64 { + t.Helper() + resource, validResource := event.Object.(tiltresource.Object) + require.True(t, validResource) + version, parseErr := strconv.ParseUint(resource.GetObjectMeta().ResourceVersion, 10, 64) + require.NoError(t, parseErr) + return version +}