From 7ba92e7389b0ce811edc34a2c223d0146e8cadf3 Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Mon, 13 Jul 2026 14:50:30 -0700 Subject: [PATCH] feat(gateway): persist request receipts on Land Create authoritative request summaries, exact change URI mappings, and queue receipt projections before publishing accepted Land requests. Validation: make fmt && make build && make test && make e2e-test --- submitqueue/core/request/BUILD.bazel | 2 + submitqueue/core/request/receipt.go | 41 +++++ submitqueue/core/request/receipt_test.go | 74 ++++++++ submitqueue/entity/request_log.go | 6 +- submitqueue/gateway/controller/BUILD.bazel | 2 + submitqueue/gateway/controller/land.go | 92 ++++++--- submitqueue/gateway/controller/land_test.go | 174 ++++++++++++++++-- submitqueue/gateway/controller/read_errors.go | 59 ++++++ .../controller/storage_fixture_test.go | 156 ++++++++++++++++ 9 files changed, 561 insertions(+), 45 deletions(-) create mode 100644 submitqueue/core/request/receipt.go create mode 100644 submitqueue/core/request/receipt_test.go create mode 100644 submitqueue/gateway/controller/read_errors.go create mode 100644 submitqueue/gateway/controller/storage_fixture_test.go diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index a8fb37ab..65abf76c 100644 --- a/submitqueue/core/request/BUILD.bazel +++ b/submitqueue/core/request/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "log.go", + "receipt.go", "request.go", ], importpath = "github.com/uber/submitqueue/submitqueue/core/request", @@ -21,6 +22,7 @@ go_test( name = "go_default_test", srcs = [ "log_test.go", + "receipt_test.go", "request_test.go", ], embed = [":go_default_library"], diff --git a/submitqueue/core/request/receipt.go b/submitqueue/core/request/receipt.go new file mode 100644 index 00000000..64b2986b --- /dev/null +++ b/submitqueue/core/request/receipt.go @@ -0,0 +1,41 @@ +// 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" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +// ReceiptWriter stores an internal request receipt before pipeline publication. +type ReceiptWriter struct { + store storage.Storage +} + +// NewReceiptWriter creates a request receipt writer. +func NewReceiptWriter(store storage.Storage) *ReceiptWriter { + return &ReceiptWriter{store: store} +} + +// Create writes the authoritative request receipt. +func (w *ReceiptWriter) Create(ctx context.Context, summary entity.RequestSummary) error { + if err := w.store.GetRequestSummaryStore().Create(ctx, summary); err != nil { + return fmt.Errorf("failed to create request summary request_id=%s: %w", summary.RequestID, err) + } + return nil +} diff --git a/submitqueue/core/request/receipt_test.go b/submitqueue/core/request/receipt_test.go new file mode 100644 index 00000000..0c96a6dd --- /dev/null +++ b/submitqueue/core/request/receipt_test.go @@ -0,0 +1,74 @@ +// 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" + "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 TestReceiptWriter_Create(t *testing.T) { + summary := testRequestSummary() + tests := []struct { + name string + setup func(*gomock.Controller, *storagemock.MockStorage) + wantError bool + }{ + { + name: "creates receipt", + 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(nil) + }, + }, + { + name: "summary failure stops remaining writes", + 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) + }, + wantError: true, + }, + } + + 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 := NewReceiptWriter(store).Create(context.Background(), summary) + if tt.wantError { + 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.RequestStatusAccepting, StatusTimestampMs: 10, Version: 1, Metadata: map[string]string{}, + } +} diff --git a/submitqueue/entity/request_log.go b/submitqueue/entity/request_log.go index 6dee8839..17899c09 100644 --- a/submitqueue/entity/request_log.go +++ b/submitqueue/entity/request_log.go @@ -31,7 +31,11 @@ const ( // RequestStatusUnknown is the unknown sentinel status. It is set by default when the structure is initialized. It should never be seen in the system. RequestStatusUnknown RequestStatus = "" - // RequestStatusAccepted indicates that the request has been accepted by the system. Typically a gateway service will set this status when the land request is received and persisted to the logging database. + // RequestStatusAccepting is the internal status of a persisted Land receipt that has not yet been published to the processing pipeline. + // Public read APIs must not expose requests that remain in this status. + RequestStatusAccepting RequestStatus = "accepting" + + // RequestStatusAccepted indicates that the request has been published to the processing pipeline. RequestStatusAccepted RequestStatus = "accepted" // RequestStatusStarted is the initial status of a request. It corresponds to the RequestStateStarted state and typically set by the orchestrator service when the request is received and persisted to the operating database. diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index 45a7c6c9..d3ba427c 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "cancel.go", "land.go", "ping.go", + "read_errors.go", "status.go", ], importpath = "github.com/uber/submitqueue/submitqueue/gateway/controller", @@ -37,6 +38,7 @@ go_test( "land_test.go", "ping_test.go", "status_test.go", + "storage_fixture_test.go", ], embed = [":go_default_library"], deps = [ diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 6e15c72f..5f265848 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" @@ -36,13 +38,15 @@ import ( "go.uber.org/zap" ) +var errInvalidRequest = errors.New("invalid request") + // ErrInvalidRequest is returned when the request fails validation. // This error should be mapped to codes.InvalidArgument at the gRPC layer. -var ErrInvalidRequest = errs.NewUserError(errors.New("invalid request")) +var ErrInvalidRequest = errs.NewUserError(errInvalidRequest) // IsInvalidRequest returns true if any error in the error chain is ErrInvalidRequest. func IsInvalidRequest(err error) bool { - return errors.Is(err, ErrInvalidRequest) + return errors.Is(err, errInvalidRequest) } // UnrecognizedQueueError indicates the request named a queue that is not @@ -65,12 +69,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 + receiptWriter *requestcore.ReceiptWriter + queueConfigs queueconfig.Store + registry consumer.TopicRegistry } // NewLandController creates a new instance of the gateway land controller. @@ -78,12 +83,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, + receiptWriter: requestcore.NewReceiptWriter(store), + queueConfigs: queueConfigs, + registry: registry, } } @@ -94,13 +100,16 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p op := metrics.Begin(c.metricsScope, opName) defer func() { op.Complete(retErr) }() - // Validate required fields. - if req.Queue == "" { - return nil, fmt.Errorf("LandController requires the request to have a queue name specified: %w", ErrInvalidRequest) + // Validate provider-agnostic request constraints before allocating an sqid. + if err := validateQueueIdentifier(req.Queue); err != nil { + return nil, fmt.Errorf("LandController invalid queue: %w", err) } - if req.Change == nil || len(req.Change.Uris) == 0 { + if req.Change == nil { return nil, fmt.Errorf("LandController requires the request to have at least one change URI specified: %w", ErrInvalidRequest) } + if err := validateChangeURIs(req.Change.Uris); err != nil { + return nil, fmt.Errorf("LandController invalid change URIs: %w", err) + } change := change.Change{ URIs: req.Change.GetUris(), @@ -132,13 +141,45 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p Change: change, LandStrategy: strategy, } + if err := validateStoredIdentifier("generated sqid", landRequest.ID); err != nil { + return nil, fmt.Errorf("LandController generated invalid request ID for queue=%s: %w", queue, err) + } + receivedAtMs := time.Now().UnixMilli() + summary := entity.RequestSummary{ + RequestID: landRequest.ID, + Queue: landRequest.Queue, + ChangeURIs: append([]string{}, landRequest.Change.URIs...), + ReceivedAtMs: receivedAtMs, + Status: entity.RequestStatusAccepting, + StatusTimestampMs: receivedAtMs, + Version: 1, + Metadata: map[string]string{}, + } + if err := c.receiptWriter.Create(ctx, summary); err != nil { + return nil, fmt.Errorf("LandController failed to create request receipt sqid=%s: %w", landRequest.ID, err) + } + + // Publish before exposing the request as accepted. A failed publish leaves an + // internal accepting receipt that public read APIs do not expose. + if err := c.publishToQueue(ctx, landRequest); err != nil { + return nil, fmt.Errorf("LandController failed to publish request to queue: %w", 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) + // Publication is the Land success boundary. Returning an error here would + // encourage the caller to submit a duplicate request that is already queued. + c.logger.Errorw("failed to record accepted status after publishing request", + "queue", req.Queue, + "sqid", landRequest.ID, + "error", err, + ) + metrics.NamedCounter(c.metricsScope, opName, "accepted_log_failure", 1) } c.logger.Debugw("land request created", @@ -149,11 +190,6 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p "strategy", string(strategy), ) - // Publish to queue for async processing - if err := c.publishToQueue(ctx, landRequest); err != nil { - return nil, fmt.Errorf("LandController failed to publish request to queue: %w", err) - } - c.logger.Infow("request published to queue", "queue", req.Queue, "sqid", landRequest.ID, diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index 63df3c11..2142fbcb 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -17,6 +17,7 @@ package controller import ( "context" "fmt" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -69,11 +70,7 @@ 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) - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - return store + return newControllerStorageFixture(ctrl).storage } // noopQueueConfigStore returns a mock queueconfig.Store that always reports @@ -169,6 +166,48 @@ func TestLand_ReturnsErrorOnEmptyQueue(t *testing.T) { assert.True(t, IsInvalidRequest(err)) } +func TestLand_ValidatesQueueLengthBeforeAllocatingSqid(t *testing.T) { + tests := []struct { + name string + queue string + wantError bool + }{ + {name: "maximum length", queue: strings.Repeat("q", maxQueueIdentifierBytes)}, + {name: "over maximum", queue: strings.Repeat("q", maxQueueIdentifierBytes+1), wantError: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + if !tt.wantError { + cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(1), nil) + } + controller := NewLandController( + zap.NewNop().Sugar(), + tally.NoopScope, + cnt, + noopStorage(ctrl), + noopQueueConfigStore(ctrl), + newTestRegistryWithNoopPublisher(t, ctrl), + ) + + resp, err := controller.Land(context.Background(), &pb.LandRequest{ + Queue: tt.queue, + Change: &changepb.Change{Uris: []string{"uri"}}, + }) + + if tt.wantError { + require.Error(t, err) + assert.True(t, IsInvalidRequest(err)) + return + } + require.NoError(t, err) + assert.Equal(t, tt.queue+"/1", resp.Sqid) + }) + } +} + func TestLand_ReturnsErrorOnEmptyChangeUri(t *testing.T) { ctrl := gomock.NewController(t) @@ -186,6 +225,39 @@ func TestLand_ReturnsErrorOnEmptyChangeUri(t *testing.T) { assert.True(t, IsInvalidRequest(err)) } +func TestLand_ReturnsErrorOnInvalidChangeURIs(t *testing.T) { + tests := []struct { + name string + uris []string + }{ + {name: "empty URI element", uris: []string{""}}, + {name: "URI exceeds storage limit", uris: []string{strings.Repeat("x", maxStorageIdentifierBytes+1)}}, + {name: "duplicate exact URI", uris: []string{"uri", "uri"}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + controller := NewLandController( + zap.NewNop().Sugar(), + tally.NoopScope, + countermock.NewMockCounter(ctrl), + noopStorage(ctrl), + noopQueueConfigStore(ctrl), + newTestRegistryWithNoopPublisher(t, ctrl), + ) + + _, err := controller.Land(context.Background(), &pb.LandRequest{ + Queue: "test-queue", + Change: &changepb.Change{Uris: tt.uris}, + }) + + require.Error(t, err) + assert.True(t, IsInvalidRequest(err)) + }) + } +} + func TestLand_ReturnsErrorOnNilChange(t *testing.T) { ctrl := gomock.NewController(t) @@ -253,22 +325,45 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { func TestLand_PublishesToQueue(t *testing.T) { var publishedTopic string var publishedMessage entityqueue.Message + var persistedSummary entity.RequestSummary + 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) + logStore := storagemock.NewMockRequestLogStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).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 + }, + ), + 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 + }, + ), + logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, log entity.RequestLog) error { + persistedLog = log + 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 +376,24 @@ 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.RequestStatusAccepting, + StatusTimestampMs: persistedSummary.ReceivedAtMs, + Version: 1, + Metadata: map[string]string{}, + }, persistedSummary) + assert.Positive(t, persistedSummary.ReceivedAtMs) + 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) @@ -295,16 +408,22 @@ func TestLand_PublishesToQueue(t *testing.T) { assert.Equal(t, mergestrategy.MergeStrategyRebase, deserializedReq.LandStrategy) } -func TestLand_ContinuesWhenPublishFails(t *testing.T) { +func TestLand_ReturnsErrorWhenPublishFails(t *testing.T) { ctrl := gomock.NewController(t) - cnt := countermock.NewMockCounter(ctrl) cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(999), nil) + store := storagemock.NewMockStorage(ctrl) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + store.EXPECT().GetRequestSummaryStore().Return(summaryStore) + summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestSummary) error { + assert.Equal(t, entity.RequestStatusAccepting, summary.Status) + return nil + }) registry, publisher := newTestRegistry(t, ctrl) publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("queue unavailable")) - 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{ @@ -313,6 +432,29 @@ func TestLand_ContinuesWhenPublishFails(t *testing.T) { } _, err := controller.Land(ctx, req) - // Should fail if publish fails require.Error(t, err) } + +func TestLand_ReturnsSqidWhenAcceptedLogFailsAfterPublish(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(999), nil) + fixture := newControllerStorageFixture(ctrl) + fixture.setLogInsertError(fmt.Errorf("log unavailable")) + + controller := NewLandController( + zap.NewNop().Sugar(), + tally.NoopScope, + cnt, + fixture.storage, + noopQueueConfigStore(ctrl), + newTestRegistryWithNoopPublisher(t, ctrl), + ) + resp, err := controller.Land(context.Background(), &pb.LandRequest{ + Queue: "test-queue", + Change: &changepb.Change{Uris: []string{"uri"}}, + }) + + require.NoError(t, err) + assert.Equal(t, "test-queue/999", resp.Sqid) +} diff --git a/submitqueue/gateway/controller/read_errors.go b/submitqueue/gateway/controller/read_errors.go new file mode 100644 index 00000000..feaa3e3a --- /dev/null +++ b/submitqueue/gateway/controller/read_errors.go @@ -0,0 +1,59 @@ +// 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 controller + +import "fmt" + +const ( + maxQueueIdentifierBytes = 235 + maxStorageIdentifierBytes = 255 +) + +func validateQueueIdentifier(queue string) error { + if queue == "" { + return fmt.Errorf("queue must be non-empty: %w", ErrInvalidRequest) + } + if len(queue) > maxQueueIdentifierBytes { + return fmt.Errorf("queue exceeds %d bytes: %w", maxQueueIdentifierBytes, ErrInvalidRequest) + } + return nil +} + +func validateStoredIdentifier(name, value string) error { + if value == "" { + return fmt.Errorf("%s must be non-empty: %w", name, ErrInvalidRequest) + } + if len(value) > maxStorageIdentifierBytes { + return fmt.Errorf("%s exceeds %d bytes: %w", name, maxStorageIdentifierBytes, ErrInvalidRequest) + } + return nil +} + +func validateChangeURIs(changeURIs []string) error { + if len(changeURIs) == 0 { + return fmt.Errorf("at least one change URI is required: %w", ErrInvalidRequest) + } + seen := make(map[string]struct{}, len(changeURIs)) + for _, changeURI := range changeURIs { + if err := validateStoredIdentifier("change URI", changeURI); err != nil { + return err + } + if _, ok := seen[changeURI]; ok { + return fmt.Errorf("duplicate change URI %q: %w", changeURI, ErrInvalidRequest) + } + seen[changeURI] = struct{}{} + } + return nil +} diff --git a/submitqueue/gateway/controller/storage_fixture_test.go b/submitqueue/gateway/controller/storage_fixture_test.go new file mode 100644 index 00000000..832aecdc --- /dev/null +++ b/submitqueue/gateway/controller/storage_fixture_test.go @@ -0,0 +1,156 @@ +// 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 controller + +import ( + "context" + "fmt" + "sync" + + "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" +) + +// controllerStorageFixture provides shared stateful storage behavior for gateway controller tests. +type controllerStorageFixture struct { + storage *storagemock.MockStorage + summaryStore *storagemock.MockRequestSummaryStore + queueStore *storagemock.MockRequestQueueSummaryStore + uriStore *storagemock.MockRequestURIStore + logStore *storagemock.MockRequestLogStore + mu sync.Mutex + summaries map[string]entity.RequestSummary + queueSummaries map[string]entity.RequestQueueSummary + logs []entity.RequestLog + logInsertErr error +} + +func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixture { + fixture := &controllerStorageFixture{ + storage: storagemock.NewMockStorage(ctrl), + summaryStore: storagemock.NewMockRequestSummaryStore(ctrl), + queueStore: storagemock.NewMockRequestQueueSummaryStore(ctrl), + uriStore: storagemock.NewMockRequestURIStore(ctrl), + logStore: storagemock.NewMockRequestLogStore(ctrl), + summaries: make(map[string]entity.RequestSummary), + queueSummaries: make(map[string]entity.RequestQueueSummary), + } + fixture.storage.EXPECT().GetRequestSummaryStore().Return(fixture.summaryStore).AnyTimes() + fixture.storage.EXPECT().GetRequestQueueSummaryStore().Return(fixture.queueStore).AnyTimes() + fixture.storage.EXPECT().GetRequestURIStore().Return(fixture.uriStore).AnyTimes() + fixture.storage.EXPECT().GetRequestLogStore().Return(fixture.logStore).AnyTimes() + + fixture.summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestSummary) error { + fixture.mu.Lock() + defer fixture.mu.Unlock() + if _, ok := fixture.summaries[summary.RequestID]; ok { + return storage.ErrAlreadyExists + } + fixture.summaries[summary.RequestID] = summary + return nil + }).AnyTimes() + fixture.summaryStore.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, requestID string) (entity.RequestSummary, error) { + fixture.mu.Lock() + defer fixture.mu.Unlock() + summary, ok := fixture.summaries[requestID] + if !ok { + return entity.RequestSummary{}, storage.ErrNotFound + } + return summary, nil + }).AnyTimes() + fixture.summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) error { + fixture.mu.Lock() + defer fixture.mu.Unlock() + current, ok := fixture.summaries[summary.RequestID] + if !ok { + return storage.ErrNotFound + } + if current.Version != oldVersion { + return storage.ErrVersionMismatch + } + summary.Version = newVersion + fixture.summaries[summary.RequestID] = summary + return nil + }).AnyTimes() + + fixture.queueStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestQueueSummary) error { + fixture.mu.Lock() + defer fixture.mu.Unlock() + key := queueSummaryTestKey(summary.Queue, summary.ReceivedAtMs, summary.RequestID) + if _, ok := fixture.queueSummaries[key]; ok { + return storage.ErrAlreadyExists + } + fixture.queueSummaries[key] = summary + return nil + }).AnyTimes() + fixture.queueStore.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, queue string, receivedAtMs int64, requestID string) (entity.RequestQueueSummary, error) { + fixture.mu.Lock() + defer fixture.mu.Unlock() + summary, ok := fixture.queueSummaries[queueSummaryTestKey(queue, receivedAtMs, requestID)] + if !ok { + return entity.RequestQueueSummary{}, storage.ErrNotFound + } + return summary, nil + }).AnyTimes() + fixture.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestQueueSummary, oldVersion, newVersion int32) error { + fixture.mu.Lock() + defer fixture.mu.Unlock() + key := queueSummaryTestKey(summary.Queue, summary.ReceivedAtMs, summary.RequestID) + current, ok := fixture.queueSummaries[key] + if !ok { + return storage.ErrNotFound + } + if current.Version != oldVersion { + return storage.ErrVersionMismatch + } + summary.Version = newVersion + fixture.queueSummaries[key] = summary + return nil + }).AnyTimes() + + fixture.uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + fixture.logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, log entity.RequestLog) error { + fixture.mu.Lock() + defer fixture.mu.Unlock() + if fixture.logInsertErr != nil { + return fixture.logInsertErr + } + fixture.logs = append(fixture.logs, log) + return nil + }).AnyTimes() + return fixture +} + +func (f *controllerStorageFixture) setLogInsertError(err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.logInsertErr = err +} + +func (f *controllerStorageFixture) addSummary(summary entity.RequestSummary) { + f.mu.Lock() + defer f.mu.Unlock() + f.summaries[summary.RequestID] = summary + f.queueSummaries[queueSummaryTestKey(summary.Queue, summary.ReceivedAtMs, summary.RequestID)] = entity.RequestQueueSummary{ + RequestID: summary.RequestID, Queue: summary.Queue, ChangeURIs: summary.ChangeURIs, ReceivedAtMs: summary.ReceivedAtMs, + Status: summary.Status, Version: summary.Version, LastError: summary.LastError, Metadata: summary.Metadata, + } +} + +func queueSummaryTestKey(queue string, receivedAtMs int64, requestID string) string { + return fmt.Sprintf("%s\x00%d\x00%s", queue, receivedAtMs, requestID) +}