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)