From a16568310f157a4adc0e76707fbc87b52083355c Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Wed, 8 Jul 2026 15:53:26 -0700 Subject: [PATCH] feat(gateway): persist request context on Land Summary: Add a focused admission writer that creates authoritative request context, change URI mappings, and the initial queue projection before Land appends its accepted log or publishes asynchronous work. Duplicate creates are reconciled in the application layer so identical retries converge and conflicting immutable context fails fast. Test Plan: make fmt make gazelle ./tool/bazel test //submitqueue/core/request:go_default_test //submitqueue/gateway/controller:go_default_test --test_output=errors Revert Plan: Revert this commit to stop writing request context. Existing additive rows may remain unused. API Changes: None. Monitoring and Alerts: Land controller operation metrics report admission write failures through the existing failure path. --- submitqueue/core/request/BUILD.bazel | 2 + submitqueue/core/request/admission.go | 122 ++++++++++++++++++++ submitqueue/core/request/admission_test.go | 114 ++++++++++++++++++ submitqueue/gateway/controller/land.go | 51 +++++--- submitqueue/gateway/controller/land_test.go | 100 ++++++++++++++-- 5 files changed, 366 insertions(+), 23 deletions(-) create mode 100644 submitqueue/core/request/admission.go create mode 100644 submitqueue/core/request/admission_test.go diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index a8fb37ab..d5777a6e 100644 --- a/submitqueue/core/request/BUILD.bazel +++ b/submitqueue/core/request/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "admission.go", "log.go", "request.go", ], @@ -20,6 +21,7 @@ go_library( go_test( name = "go_default_test", srcs = [ + "admission_test.go", "log_test.go", "request_test.go", ], diff --git a/submitqueue/core/request/admission.go b/submitqueue/core/request/admission.go new file mode 100644 index 00000000..8194bf01 --- /dev/null +++ b/submitqueue/core/request/admission.go @@ -0,0 +1,122 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package request + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +// AdmissionWriter creates immutable request context and initial read-model projections. +// Storage implementations remain mechanical; this type decides whether duplicate creates are identical retries or conflicts. +type AdmissionWriter struct { + store storage.Storage +} + +// NewAdmissionWriter creates a request receipt projection writer. +func NewAdmissionWriter(store storage.Storage) *AdmissionWriter { + return &AdmissionWriter{store: store} +} + +// Create writes immutable request context and initial accepted projections. +// A duplicate for the same request ID is accepted only when its immutable context matches exactly. +func (m *AdmissionWriter) Create(ctx context.Context, summary entity.RequestSummary) error { + if err := m.createRequestSummary(ctx, summary); err != nil { + return err + } + + for _, changeURI := range summary.ChangeURIs { + mapping := entity.RequestURI{ + ChangeURI: changeURI, + ReceivedAtMs: summary.ReceivedAtMs, + RequestID: summary.RequestID, + } + if err := m.store.GetRequestURIStore().Create(ctx, mapping); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("failed to create request URI mapping request_id=%s change_uri=%s: %w", summary.RequestID, changeURI, err) + } + } + + queueSummary := queueSummaryFromSummary(summary) + if err := m.store.GetRequestQueueSummaryStore().Create(ctx, queueSummary); err != nil { + if !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("failed to create queue summary request_id=%s: %w", summary.RequestID, err) + } + existing, getErr := m.store.GetRequestQueueSummaryStore().Get(ctx, summary.Queue, summary.ReceivedAtMs, summary.RequestID) + if getErr != nil { + return fmt.Errorf("failed to get duplicate queue summary request_id=%s: %w", summary.RequestID, getErr) + } + if !sameQueueSummaryIdentity(existing, queueSummary) { + return fmt.Errorf("conflicting queue summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists) + } + } + + return nil +} + +func (m *AdmissionWriter) createRequestSummary(ctx context.Context, summary entity.RequestSummary) error { + if err := m.store.GetRequestSummaryStore().Create(ctx, summary); err != nil { + if !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("failed to create request summary request_id=%s: %w", summary.RequestID, err) + } + existing, getErr := m.store.GetRequestSummaryStore().Get(ctx, summary.RequestID) + if getErr != nil { + return fmt.Errorf("failed to get duplicate request summary request_id=%s: %w", summary.RequestID, getErr) + } + if !sameRequestSummaryIdentity(existing, summary) { + return fmt.Errorf("conflicting request summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists) + } + } + return nil +} + +func queueSummaryFromSummary(summary entity.RequestSummary) entity.RequestQueueSummary { + return entity.RequestQueueSummary{ + RequestID: summary.RequestID, + Queue: summary.Queue, + ChangeURIs: slices.Clone(summary.ChangeURIs), + ReceivedAtMs: summary.ReceivedAtMs, + Status: summary.Status, + Version: summary.Version, + LastError: summary.LastError, + Metadata: cloneMetadata(summary.Metadata), + } +} + +func sameRequestSummaryIdentity(left, right entity.RequestSummary) bool { + return left.RequestID == right.RequestID && + left.Queue == right.Queue && + left.ReceivedAtMs == right.ReceivedAtMs && + slices.Equal(left.ChangeURIs, right.ChangeURIs) +} + +func sameQueueSummaryIdentity(left, right entity.RequestQueueSummary) bool { + return left.RequestID == right.RequestID && + left.Queue == right.Queue && + left.ReceivedAtMs == right.ReceivedAtMs && + slices.Equal(left.ChangeURIs, right.ChangeURIs) +} + +func cloneMetadata(metadata map[string]string) map[string]string { + if metadata == nil { + return map[string]string{} + } + return maps.Clone(metadata) +} diff --git a/submitqueue/core/request/admission_test.go b/submitqueue/core/request/admission_test.go new file mode 100644 index 00000000..8dd9ec5c --- /dev/null +++ b/submitqueue/core/request/admission_test.go @@ -0,0 +1,114 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package request + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +func TestAdmissionWriter_Create(t *testing.T) { + summary := testRequestSummary() + tests := []struct { + name string + setup func(*gomock.Controller, *storagemock.MockStorage) + wantErr error + }{ + { + name: "creates all projections", + setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) { + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() + store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() + summaryStore.EXPECT().Create(gomock.Any(), summary).Return(nil) + uriStore.EXPECT().Create(gomock.Any(), entity.RequestURI{ChangeURI: "uri/1", ReceivedAtMs: 10, RequestID: "q/1"}).Return(nil) + uriStore.EXPECT().Create(gomock.Any(), entity.RequestURI{ChangeURI: "uri/2", ReceivedAtMs: 10, RequestID: "q/1"}).Return(nil) + queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(summary)).Return(nil) + }, + }, + { + name: "identical retry succeeds", + setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) { + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() + store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() + summaryStore.EXPECT().Create(gomock.Any(), summary).Return(storage.ErrAlreadyExists) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(summary, nil) + uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists).Times(2) + queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(summary)).Return(storage.ErrAlreadyExists) + queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(summary), nil) + }, + }, + { + name: "conflicting summary retry fails", + setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) { + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + summaryStore.EXPECT().Create(gomock.Any(), summary).Return(storage.ErrAlreadyExists) + conflict := summary + conflict.Queue = "other" + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(conflict, nil) + }, + wantErr: storage.ErrAlreadyExists, + }, + { + name: "URI write failure stops queue projection", + setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) { + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() + summaryStore.EXPECT().Create(gomock.Any(), summary).Return(nil) + uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(fmt.Errorf("URI down")) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + store := storagemock.NewMockStorage(ctrl) + tt.setup(ctrl, store) + err := NewAdmissionWriter(store).Create(context.Background(), summary) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else if tt.name == "URI write failure stops queue projection" { + require.Error(t, err) + } else { + require.NoError(t, err) + } + }) + } +} + +func testRequestSummary() entity.RequestSummary { + return entity.RequestSummary{ + RequestID: "q/1", Queue: "q", ChangeURIs: []string{"uri/1", "uri/2"}, ReceivedAtMs: 10, + Status: entity.RequestStatusAccepted, StatusTimestampMs: 10, Version: 1, Metadata: map[string]string{}, + } +} diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 6e15c72f..c4ef987f 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/uber-go/tally" mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" @@ -29,6 +30,7 @@ import ( "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/extension/counter" "github.com/uber/submitqueue/platform/metrics" + requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" @@ -65,12 +67,13 @@ func IsUnrecognizedQueue(err error) bool { // LandController handles land business logic for the gateway type LandController struct { - logger *zap.SugaredLogger - metricsScope tally.Scope - counter counter.Counter - store storage.Storage - queueConfigs queueconfig.Store - registry consumer.TopicRegistry + logger *zap.SugaredLogger + metricsScope tally.Scope + counter counter.Counter + store storage.Storage + admissionWriter *requestcore.AdmissionWriter + queueConfigs queueconfig.Store + registry consumer.TopicRegistry } // NewLandController creates a new instance of the gateway land controller. @@ -78,12 +81,13 @@ type LandController struct { // topickey.TopicKeyStart in the registry. func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter counter.Counter, store storage.Storage, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) *LandController { return &LandController{ - logger: logger, - metricsScope: scope.SubScope("land_controller"), - counter: counter, - store: store, - queueConfigs: queueConfigs, - registry: registry, + logger: logger, + metricsScope: scope.SubScope("land_controller"), + counter: counter, + store: store, + admissionWriter: requestcore.NewAdmissionWriter(store), + queueConfigs: queueConfigs, + registry: registry, } } @@ -132,13 +136,32 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p Change: change, LandStrategy: strategy, } + receivedAtMs := time.Now().UnixMilli() + summary := entity.RequestSummary{ + RequestID: landRequest.ID, + Queue: landRequest.Queue, + ChangeURIs: append([]string{}, landRequest.Change.URIs...), + ReceivedAtMs: receivedAtMs, + Status: entity.RequestStatusAccepted, + StatusTimestampMs: receivedAtMs, + Version: 1, + Metadata: map[string]string{}, + } + if err := c.admissionWriter.Create(ctx, summary); err != nil { + return nil, fmt.Errorf("LandController failed to create request receipt sqid=%s: %w", landRequest.ID, err) + } // Record the accepted status in the request log for reconciliation. Once the request materializes as a Request entity, the status might be updated to "new". // It is important to record the status before publishing to the queue for processing. It is important to publish straight to the database and not via a entityqueue. // Gateway has to stay consistent with the request log. - logEntry := entity.NewRequestLog(landRequest.ID, entity.RequestStatusAccepted, 0, "", nil) + logEntry := entity.RequestLog{ + RequestID: landRequest.ID, + TimestampMs: receivedAtMs, + Status: entity.RequestStatusAccepted, + Metadata: map[string]string{}, + } if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil { - return nil, fmt.Errorf("LandController failed to insert request log for sqid=%s: %w", landRequest.ID, err) + return nil, fmt.Errorf("LandController failed to insert accepted request log for sqid=%s: %w", landRequest.ID, err) } c.logger.Debugw("land request created", diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index 63df3c11..cf0e2081 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -69,10 +69,19 @@ func newTestRegistryWithNoopPublisher(t *testing.T, ctrl *gomock.Controller) con // noopStorage returns a storage.Storage whose RequestLogStore.Insert // succeeds silently for any entityqueue. func noopStorage(ctrl *gomock.Controller) storage.Storage { - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() store := storagemock.NewMockStorage(ctrl) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) + logStore := storagemock.NewMockRequestLogStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() + store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + queueStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() return store } @@ -253,22 +262,63 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { func TestLand_PublishesToQueue(t *testing.T) { var publishedTopic string var publishedMessage entityqueue.Message + var persistedSummary entity.RequestSummary + var persistedMapping entity.RequestURI + var persistedQueueSummary entity.RequestQueueSummary + var persistedLog entity.RequestLog ctrl := gomock.NewController(t) cnt := countermock.NewMockCounter(ctrl) cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(123), nil) + store := storagemock.NewMockStorage(ctrl) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) + logStore := storagemock.NewMockRequestLogStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() + store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() + store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + registry, publisher := newTestRegistry(t, ctrl) - publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, topic string, msg entityqueue.Message) error { - publishedTopic = topic - publishedMessage = msg - return nil - }, + + gomock.InOrder( + summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, summary entity.RequestSummary) error { + persistedSummary = summary + return nil + }, + ), + uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, mapping entity.RequestURI) error { + persistedMapping = mapping + return nil + }, + ), + queueStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, summary entity.RequestQueueSummary) error { + persistedQueueSummary = summary + return nil + }, + ), + logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, log entity.RequestLog) error { + persistedLog = log + return nil + }, + ), + publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, topic string, msg entityqueue.Message) error { + publishedTopic = topic + publishedMessage = msg + return nil + }, + ), ) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, cnt, noopStorage(ctrl), noopQueueConfigStore(ctrl), registry) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, cnt, store, noopQueueConfigStore(ctrl), registry) ctx := context.Background() req := &pb.LandRequest{ @@ -281,6 +331,38 @@ func TestLand_PublishesToQueue(t *testing.T) { require.NoError(t, err) assert.Equal(t, "test-queue/123", resp.Sqid) + assert.Equal(t, entity.RequestSummary{ + RequestID: "test-queue/123", + Queue: "test-queue", + ChangeURIs: []string{"github://github.example.com/uber/backend/pull/456/fedcba9876543210fedcba9876543210fedcba98"}, + ReceivedAtMs: persistedSummary.ReceivedAtMs, + Status: entity.RequestStatusAccepted, + StatusTimestampMs: persistedSummary.ReceivedAtMs, + Version: 1, + Metadata: map[string]string{}, + }, persistedSummary) + assert.Positive(t, persistedSummary.ReceivedAtMs) + assert.Equal(t, entity.RequestURI{ + ChangeURI: "github://github.example.com/uber/backend/pull/456/fedcba9876543210fedcba9876543210fedcba98", + ReceivedAtMs: persistedSummary.ReceivedAtMs, + RequestID: "test-queue/123", + }, persistedMapping) + assert.Equal(t, entity.RequestQueueSummary{ + RequestID: "test-queue/123", + Queue: "test-queue", + ChangeURIs: []string{"github://github.example.com/uber/backend/pull/456/fedcba9876543210fedcba9876543210fedcba98"}, + ReceivedAtMs: persistedSummary.ReceivedAtMs, + Status: entity.RequestStatusAccepted, + Version: 1, + Metadata: map[string]string{}, + }, persistedQueueSummary) + assert.Equal(t, entity.RequestLog{ + RequestID: "test-queue/123", + TimestampMs: persistedSummary.ReceivedAtMs, + Status: entity.RequestStatusAccepted, + Metadata: map[string]string{}, + }, persistedLog) + // Verify message was published to the topic registered under TopicKeyStart assert.Equal(t, "start", publishedTopic) assert.Equal(t, "test-queue/123", publishedMessage.ID)