From 5fc49f1a4938e55c589768b159035ac7f2f1e078 Mon Sep 17 00:00:00 2001 From: manjari Date: Tue, 13 Oct 2026 19:59:45 +0000 Subject: [PATCH 1/5] feat(gateway): Reject land requests for a paused queue --- service/submitqueue/README.md | 1 + .../submitqueue/gateway/server/BUILD.bazel | 3 + service/submitqueue/gateway/server/main.go | 27 +++++- .../submitqueue/gateway/server/main_test.go | 5 ++ submitqueue/gateway/controller/BUILD.bazel | 3 + submitqueue/gateway/controller/land.go | 30 ++++++- submitqueue/gateway/controller/land_test.go | 87 +++++++++++++++++-- 7 files changed, 149 insertions(+), 7 deletions(-) diff --git a/service/submitqueue/README.md b/service/submitqueue/README.md index 365a77947..e35933c5a 100644 --- a/service/submitqueue/README.md +++ b/service/submitqueue/README.md @@ -21,6 +21,7 @@ Each service requires the application and queue MySQL databases. Service-specifi | `MYSQL_DSN` | both | yes | Application database DSN | — | | `QUEUE_MYSQL_DSN` | both | yes | Queue database DSN | — | | `QUEUE_CONFIG_PATH` | Gateway | yes | YAML registry of accepted queue names | — | +| `QUEUE_PAUSE_CONFIG_PATH` | Gateway | no | YAML file of per-queue pause switches, re-read on every land | pausing disabled when unset | | `PROFILES_CONFIG_PATH` | Orchestrator | no | YAML extension profiles and per-queue routing | built-in example profiles | | `HOSTNAME` | both | no | Base subscriber name; Gateway appends its PID | time-based service name | | `QUEUE_LOG_LEVEL` | both | no | Message-queue logger level | `info` | diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 28b4b4364..065125874 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -32,6 +32,9 @@ go_library( "//service/submitqueue/gateway/server/mapper:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/extension/queueconfig/yaml:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + "//submitqueue/extension/queuepause/file:go_default_library", + "//submitqueue/extension/queuepause/noop:go_default_library", "//submitqueue/gateway/controller:go_default_library", "//submitqueue/gateway/controller/log:go_default_library", "//submitqueue/gateway/core/request:go_default_library", diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 88611a65b..e30b6eb5a 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -44,6 +44,9 @@ import ( "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper" "github.com/uber/submitqueue/submitqueue/core/topickey" yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" + queuepausefile "github.com/uber/submitqueue/submitqueue/extension/queuepause/file" + queuepausenoop "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop" "github.com/uber/submitqueue/submitqueue/gateway/controller" logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" @@ -159,11 +162,29 @@ func gatewayStatusError(err error) error { return status.Error(codes.Internal, err.Error()) case controller.IsInvalidRequest(err), controller.IsUnrecognizedQueue(err): return status.Error(codes.InvalidArgument, err.Error()) + case controller.IsQueuePaused(err): + return status.Error(codes.FailedPrecondition, err.Error()) default: return err } } +// newQueuePauseStore enables the file-backed pause store only when +// QUEUE_PAUSE_CONFIG_PATH is set; otherwise no queue is ever paused. +func newQueuePauseStore(logger *zap.Logger) (queuepause.Store, error) { + path := os.Getenv("QUEUE_PAUSE_CONFIG_PATH") + if path == "" { + logger.Info("queue pause disabled") + return queuepausenoop.New(), nil + } + store, err := queuepausefile.NewStore(path) + if err != nil { + return nil, fmt.Errorf("failed to load queue pause config: %w", err) + } + logger.Info("queue pause configured", zap.String("path", path)) + return store, nil +} + func main() { code := 0 if err := run(); err != nil { @@ -264,6 +285,10 @@ func run() error { if err != nil { return fmt.Errorf("failed to load queue configs: %w", err) } + queuePauses, err := newQueuePauseStore(logger) + if err != nil { + return err + } configuredQueues, err := queueConfigs.List(ctx) if err != nil { return fmt.Errorf("failed to list queue configs: %w", err) @@ -359,7 +384,7 @@ func run() error { storageFty := storageFactory{backend: store} materializer := requestcore.NewMaterializer(storageFty) pingController := controller.NewPingController(logger, scope) - landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, registry) + landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, queuePauses, registry) cancelController := controller.NewCancelController(logger.Sugar(), scope, storageFty, materializer, registry) requestSummaryController := controller.NewRequestSummaryController(logger.Sugar(), scope, storageFty) listController := controller.NewListController(logger.Sugar(), scope, storageFty, queueConfigs) diff --git a/service/submitqueue/gateway/server/main_test.go b/service/submitqueue/gateway/server/main_test.go index 2fc6cce17..c88e38c86 100644 --- a/service/submitqueue/gateway/server/main_test.go +++ b/service/submitqueue/gateway/server/main_test.go @@ -105,6 +105,11 @@ func TestGatewayStatusError(t *testing.T) { err: &controller.UnrecognizedQueueError{Queue: "missing"}, code: codes.InvalidArgument, }, + { + name: "queue paused", + err: &controller.QueuePausedError{Queue: "paused"}, + code: codes.FailedPrecondition, + }, { name: "request not found", err: &controller.RequestNotFoundError{Sqid: "1"}, diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index a56cdec6b..f3371a715 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -26,6 +26,7 @@ go_library( "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/gateway/core/request:go_default_library", "//submitqueue/gateway/extension/storage:go_default_library", @@ -63,6 +64,8 @@ go_test( "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig:go_default_library", "//submitqueue/extension/queueconfig/mock:go_default_library", + "//submitqueue/extension/queuepause/mock:go_default_library", + "//submitqueue/extension/queuepause/noop:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", "//submitqueue/gateway/core/request:go_default_library", diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 4819b0893..77ed3f465 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -31,6 +31,7 @@ import ( "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "go.uber.org/zap" @@ -63,6 +64,22 @@ func IsUnrecognizedQueue(err error) bool { return errors.As(err, &target) } +// QueuePausedError indicates the queue is refusing new land requests. +type QueuePausedError struct { + Queue string +} + +// Error implements the error interface. +func (e *QueuePausedError) Error() string { + return fmt.Sprintf("queue %q is paused and not accepting land requests", e.Queue) +} + +// IsQueuePaused returns true if any error in the chain is a *QueuePausedError. +func IsQueuePaused(err error) bool { + var target *QueuePausedError + return errors.As(err, &target) +} + // counterDomainRequest names the per-queue sequence that mints request IDs. const counterDomainRequest = "request" @@ -80,13 +97,14 @@ type landController struct { stores storage.Factory materializer *requestcore.Materializer queueConfigs queueconfig.Store + queuePauses queuepause.Store registry consumer.TopicRegistry } // NewLandController creates a new instance of the gateway land controller. // The controller publishes land requests to the topic registered under // topickey.TopicKeyStart in the registry. -func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters counter.Factory, stores storage.Factory, materializer *requestcore.Materializer, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) LandController { +func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters counter.Factory, stores storage.Factory, materializer *requestcore.Materializer, queueConfigs queueconfig.Store, queuePauses queuepause.Store, registry consumer.TopicRegistry) LandController { return &landController{ logger: logger, metricsScope: scope.SubScope("land_controller"), @@ -94,6 +112,7 @@ func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters co stores: stores, materializer: materializer, queueConfigs: queueConfigs, + queuePauses: queuePauses, registry: registry, } } @@ -121,6 +140,15 @@ func (c *landController) Land(ctx context.Context, req entity.LandRequest) (resu return entity.LandResult{}, fmt.Errorf("failed to look up queue %q: %w", queue, err) } + // A failed read fails the land: an unreadable pause state must not let a paused queue accept work. + pauseState, err := c.queuePauses.Get(ctx, queue) + if err != nil { + return entity.LandResult{}, fmt.Errorf("failed to read pause state for queue=%s: %w", queue, err) + } + if pauseState.RejectLands { + return entity.LandResult{}, errs.NewUserError(&QueuePausedError{Queue: queue}) + } + // Generate a queue-scoped request ID for the land request. // The inbound entity arrives with an empty ID; the controller owns minting it. stores, err := c.stores.For(storage.Config{QueueName: queue}) diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index d1157ba1a..7cb572d97 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -36,6 +36,8 @@ import ( "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" + queuepausemock "github.com/uber/submitqueue/submitqueue/extension/queuepause/mock" + queuepausenoop "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" @@ -87,7 +89,7 @@ func (f staticCounterFactory) For(config counter.Config) (counter.Counter, error // noopStorage returns stateful request storage whose writes succeed. func newNoopLandController(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter) LandController { fixture := newControllerStorageFixture(ctrl) - return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), newTestRegistryWithNoopPublisher(t, ctrl)) + return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl)) } // noopQueueConfigStore returns a mock queueconfig.Store that always reports @@ -178,6 +180,7 @@ func TestLand_ResolvesCounterForRequestQueue(t *testing.T) { fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), + queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl), ) ctx := context.Background() @@ -307,7 +310,7 @@ func TestLand_ReturnsUnrecognizedQueueWhenStoreReportsNotFound(t *testing.T) { qcs.EXPECT().Get(gomock.Any(), "missing-queue").Return(entity.QueueConfig{}, queueconfig.ErrNotFound) fixture := newControllerStorageFixture(ctrl) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, newTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl)) ctx := context.Background() _, err := controller.Land(ctx, testLandRequest("missing-queue")) @@ -330,7 +333,7 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { qcs.EXPECT().Get(gomock.Any(), "test-queue").Return(entity.QueueConfig{}, fmt.Errorf("config backend down")) fixture := newControllerStorageFixture(ctrl) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, newTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl)) ctx := context.Background() _, err := controller.Land(ctx, testLandRequest("test-queue")) @@ -340,6 +343,79 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { assert.False(t, IsInvalidRequest(err)) } +func newLandControllerWithPauseStore(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter, qcs *qcmock.MockStore, pauses *queuepausemock.MockStore) LandController { + fixture := newControllerStorageFixture(ctrl) + return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, pauses, newTestRegistryWithNoopPublisher(t, ctrl)) +} + +func TestLand_QueuePauseState(t *testing.T) { + tests := []struct { + name string + state entity.QueuePauseState + wantLanded bool + }{ + {name: "not paused", state: entity.QueuePauseState{}, wantLanded: true}, + {name: "freeze only still accepts lands", state: entity.QueuePauseState{FreezeProcessing: true}, wantLanded: true}, + {name: "reject only", state: entity.QueuePauseState{RejectLands: true}, wantLanded: false}, + {name: "reject and freeze", state: entity.QueuePauseState{RejectLands: true, FreezeProcessing: true}, wantLanded: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + if tt.wantLanded { + cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(1), nil) + } + pauses := queuepausemock.NewMockStore(ctrl) + pauses.EXPECT().Get(gomock.Any(), "test-queue").Return(tt.state, nil) + controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses) + + result, err := controller.Land(context.Background(), testLandRequest("test-queue")) + + if tt.wantLanded { + require.NoError(t, err) + assert.Equal(t, "1", result.ID) + return + } + require.Error(t, err) + assert.True(t, IsQueuePaused(err)) + assert.True(t, errs.IsUserError(err)) + assert.False(t, errs.IsRetryable(err)) + var typed *QueuePausedError + require.ErrorAs(t, err, &typed) + assert.Equal(t, "test-queue", typed.Queue) + }) + } +} + +func TestLand_FailsClosedWhenPauseStoreErrors(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + pauses := queuepausemock.NewMockStore(ctrl) + pauses.EXPECT().Get(gomock.Any(), "test-queue").Return(entity.QueuePauseState{}, fmt.Errorf("pause backend down")) + controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses) + + _, err := controller.Land(context.Background(), testLandRequest("test-queue")) + + require.Error(t, err) + assert.False(t, IsQueuePaused(err)) + assert.False(t, errs.IsUserError(err)) +} + +func TestLand_ReportsUnrecognizedQueueBeforeReadingPauseState(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + qcs := qcmock.NewMockStore(ctrl) + qcs.EXPECT().Get(gomock.Any(), "missing-queue").Return(entity.QueueConfig{}, queueconfig.ErrNotFound) + pauses := queuepausemock.NewMockStore(ctrl) + controller := newLandControllerWithPauseStore(t, ctrl, cnt, qcs, pauses) + + _, err := controller.Land(context.Background(), testLandRequest("missing-queue")) + + require.Error(t, err) + assert.True(t, IsUnrecognizedQueue(err)) +} + func TestLand_PublishesToQueue(t *testing.T) { var publishedTopic string var publishedMessage entityqueue.Message @@ -414,7 +490,7 @@ func TestLand_PublishesToQueue(t *testing.T) { ), ) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), registry) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), queuepausenoop.New(), registry) ctx := context.Background() req := entity.LandRequest{ @@ -486,7 +562,7 @@ func TestLand_ReturnsErrorWhenPublishFails(t *testing.T) { 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, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, storageWithSummaryStore(ctrl, summaryStore)), newControllerStorageFixture(ctrl).newMaterializer(ctrl), noopQueueConfigStore(ctrl), registry) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, storageWithSummaryStore(ctrl, summaryStore)), newControllerStorageFixture(ctrl).newMaterializer(ctrl), noopQueueConfigStore(ctrl), queuepausenoop.New(), registry) ctx := context.Background() _, err := controller.Land(ctx, testLandRequest("test-queue")) @@ -508,6 +584,7 @@ func TestLand_ReturnsSqidWhenAcceptedLogFailsAfterPublish(t *testing.T) { fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), + queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl), ) result, err := controller.Land(context.Background(), testLandRequest("test-queue")) From 3d655b8c300c0ea78cf27f0f664800f70952c38f Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 19:15:30 +0000 Subject: [PATCH 2/5] fix(gateway): Map a paused queue to FailedPrecondition --- service/submitqueue/gateway/server/BUILD.bazel | 1 + service/submitqueue/gateway/server/main.go | 6 ++++-- service/submitqueue/gateway/server/main_test.go | 5 +++-- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 065125874..ba0697afc 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -89,6 +89,7 @@ go_test( embed = [":gateway_lib"], # keep deps = [ "//api/submitqueue/gateway/protopb:go_default_library", + "//platform/errs:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig/mock:go_default_library", "//submitqueue/gateway/controller:go_default_library", diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index e30b6eb5a..26c4a9e6a 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -152,6 +152,8 @@ func (s *GatewayServer) GetRequestHistoryByChangeURI(ctx context.Context, req *p return &pb.GetRequestHistoryByChangeURIResponse{Histories: mapper.RequestHistoriesToProto(histories)}, nil } +// gatewayStatusError maps controller errors to status codes. IsInvalidRequest +// matches every user error, so each specific user error is checked before it. func gatewayStatusError(err error) error { switch { case controller.IsRequestNotFound(err): @@ -160,10 +162,10 @@ func gatewayStatusError(err error) error { return status.Error(codes.ResourceExhausted, err.Error()) case controller.IsInternalConsistency(err): return status.Error(codes.Internal, err.Error()) - case controller.IsInvalidRequest(err), controller.IsUnrecognizedQueue(err): - return status.Error(codes.InvalidArgument, err.Error()) case controller.IsQueuePaused(err): return status.Error(codes.FailedPrecondition, err.Error()) + case controller.IsInvalidRequest(err), controller.IsUnrecognizedQueue(err): + return status.Error(codes.InvalidArgument, err.Error()) default: return err } diff --git a/service/submitqueue/gateway/server/main_test.go b/service/submitqueue/gateway/server/main_test.go index c88e38c86..612fd42d5 100644 --- a/service/submitqueue/gateway/server/main_test.go +++ b/service/submitqueue/gateway/server/main_test.go @@ -23,6 +23,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/submitqueue/entity" qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" "github.com/uber/submitqueue/submitqueue/gateway/controller" @@ -102,12 +103,12 @@ func TestGatewayStatusError(t *testing.T) { }, { name: "unrecognized queue", - err: &controller.UnrecognizedQueueError{Queue: "missing"}, + err: errs.NewUserError(&controller.UnrecognizedQueueError{Queue: "missing"}), code: codes.InvalidArgument, }, { name: "queue paused", - err: &controller.QueuePausedError{Queue: "paused"}, + err: errs.NewUserError(&controller.QueuePausedError{Queue: "paused"}), code: codes.FailedPrecondition, }, { From 7bb162cf9e7be088442d70bb7eaca94963aab2cd Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:27:04 +0000 Subject: [PATCH 3/5] refactor(gateway): Use the renamed QueuePause entity in land tests --- submitqueue/gateway/controller/land_test.go | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index 7cb572d97..37d23286f 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -348,16 +348,16 @@ func newLandControllerWithPauseStore(t *testing.T, ctrl *gomock.Controller, cnt return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, pauses, newTestRegistryWithNoopPublisher(t, ctrl)) } -func TestLand_QueuePauseState(t *testing.T) { +func TestLand_QueuePause(t *testing.T) { tests := []struct { name string - state entity.QueuePauseState + state entity.QueuePause wantLanded bool }{ - {name: "not paused", state: entity.QueuePauseState{}, wantLanded: true}, - {name: "freeze only still accepts lands", state: entity.QueuePauseState{FreezeProcessing: true}, wantLanded: true}, - {name: "reject only", state: entity.QueuePauseState{RejectLands: true}, wantLanded: false}, - {name: "reject and freeze", state: entity.QueuePauseState{RejectLands: true, FreezeProcessing: true}, wantLanded: false}, + {name: "not paused", state: entity.QueuePause{}, wantLanded: true}, + {name: "freeze only still accepts lands", state: entity.QueuePause{FreezeProcessing: true}, wantLanded: true}, + {name: "reject only", state: entity.QueuePause{RejectLands: true}, wantLanded: false}, + {name: "reject and freeze", state: entity.QueuePause{RejectLands: true, FreezeProcessing: true}, wantLanded: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -392,7 +392,7 @@ func TestLand_FailsClosedWhenPauseStoreErrors(t *testing.T) { ctrl := gomock.NewController(t) cnt := countermock.NewMockCounter(ctrl) pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "test-queue").Return(entity.QueuePauseState{}, fmt.Errorf("pause backend down")) + pauses.EXPECT().Get(gomock.Any(), "test-queue").Return(entity.QueuePause{}, fmt.Errorf("pause backend down")) controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses) _, err := controller.Land(context.Background(), testLandRequest("test-queue")) From 1e33c048f67ab40b9925626f09e91becd3a23734 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:29:47 +0000 Subject: [PATCH 4/5] test(gateway): Simulate queue pause state with the fake store --- submitqueue/gateway/controller/BUILD.bazel | 3 ++- submitqueue/gateway/controller/land_test.go | 13 ++++++------- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index f3371a715..53f91cc4d 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -64,7 +64,8 @@ go_test( "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig:go_default_library", "//submitqueue/extension/queueconfig/mock:go_default_library", - "//submitqueue/extension/queuepause/mock:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + "//submitqueue/extension/queuepause/fake:go_default_library", "//submitqueue/extension/queuepause/noop:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index 37d23286f..d7fe39b3c 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -36,7 +36,8 @@ import ( "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" - queuepausemock "github.com/uber/submitqueue/submitqueue/extension/queuepause/mock" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" + queuepausefake "github.com/uber/submitqueue/submitqueue/extension/queuepause/fake" queuepausenoop "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" @@ -343,7 +344,7 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { assert.False(t, IsInvalidRequest(err)) } -func newLandControllerWithPauseStore(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter, qcs *qcmock.MockStore, pauses *queuepausemock.MockStore) LandController { +func newLandControllerWithPauseStore(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter, qcs *qcmock.MockStore, pauses queuepause.Store) LandController { fixture := newControllerStorageFixture(ctrl) return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, pauses, newTestRegistryWithNoopPublisher(t, ctrl)) } @@ -366,8 +367,7 @@ func TestLand_QueuePause(t *testing.T) { if tt.wantLanded { cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(1), nil) } - pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "test-queue").Return(tt.state, nil) + pauses := queuepausefake.New().Set("test-queue", tt.state) controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses) result, err := controller.Land(context.Background(), testLandRequest("test-queue")) @@ -391,8 +391,7 @@ func TestLand_QueuePause(t *testing.T) { func TestLand_FailsClosedWhenPauseStoreErrors(t *testing.T) { ctrl := gomock.NewController(t) cnt := countermock.NewMockCounter(ctrl) - pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "test-queue").Return(entity.QueuePause{}, fmt.Errorf("pause backend down")) + pauses := queuepausefake.New().FailWith(fmt.Errorf("pause backend down")) controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses) _, err := controller.Land(context.Background(), testLandRequest("test-queue")) @@ -407,7 +406,7 @@ func TestLand_ReportsUnrecognizedQueueBeforeReadingPauseState(t *testing.T) { cnt := countermock.NewMockCounter(ctrl) qcs := qcmock.NewMockStore(ctrl) qcs.EXPECT().Get(gomock.Any(), "missing-queue").Return(entity.QueueConfig{}, queueconfig.ErrNotFound) - pauses := queuepausemock.NewMockStore(ctrl) + pauses := queuepausefake.New().FailWith(fmt.Errorf("pause backend down")) controller := newLandControllerWithPauseStore(t, ctrl, cnt, qcs, pauses) _, err := controller.Land(context.Background(), testLandRequest("missing-queue")) From a05196403258f2936cd4dcb05b64208221372849 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:30:52 +0000 Subject: [PATCH 5/5] feat(gateway): Include the pause reason in a rejected land --- submitqueue/gateway/controller/land.go | 10 ++++++++-- submitqueue/gateway/controller/land_test.go | 4 ++++ 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 77ed3f465..11b4eaa58 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -67,11 +67,17 @@ func IsUnrecognizedQueue(err error) bool { // QueuePausedError indicates the queue is refusing new land requests. type QueuePausedError struct { Queue string + // Reason is the operator's explanation for the pause, empty when none was given. + Reason string } // Error implements the error interface. func (e *QueuePausedError) Error() string { - return fmt.Sprintf("queue %q is paused and not accepting land requests", e.Queue) + msg := fmt.Sprintf("queue %q is paused and not accepting land requests", e.Queue) + if e.Reason != "" { + msg += ": " + e.Reason + } + return msg } // IsQueuePaused returns true if any error in the chain is a *QueuePausedError. @@ -146,7 +152,7 @@ func (c *landController) Land(ctx context.Context, req entity.LandRequest) (resu return entity.LandResult{}, fmt.Errorf("failed to read pause state for queue=%s: %w", queue, err) } if pauseState.RejectLands { - return entity.LandResult{}, errs.NewUserError(&QueuePausedError{Queue: queue}) + return entity.LandResult{}, errs.NewUserError(&QueuePausedError{Queue: queue, Reason: pauseState.Reason}) } // Generate a queue-scoped request ID for the land request. diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index d7fe39b3c..6f31bf913 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -354,11 +354,14 @@ func TestLand_QueuePause(t *testing.T) { name string state entity.QueuePause wantLanded bool + wantReason string }{ {name: "not paused", state: entity.QueuePause{}, wantLanded: true}, {name: "freeze only still accepts lands", state: entity.QueuePause{FreezeProcessing: true}, wantLanded: true}, {name: "reject only", state: entity.QueuePause{RejectLands: true}, wantLanded: false}, {name: "reject and freeze", state: entity.QueuePause{RejectLands: true, FreezeProcessing: true}, wantLanded: false}, + {name: "reject carries the reason", state: entity.QueuePause{RejectLands: true, Reason: "repo freeze until 17:00"}, wantLanded: false, wantReason: "repo freeze until 17:00"}, + {name: "reason alone does not reject", state: entity.QueuePause{Reason: "stale note"}, wantLanded: true}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -384,6 +387,7 @@ func TestLand_QueuePause(t *testing.T) { var typed *QueuePausedError require.ErrorAs(t, err, &typed) assert.Equal(t, "test-queue", typed.Queue) + assert.Equal(t, tt.wantReason, typed.Reason) }) } }