From 837041dd31c50f4faaa2b9758ff9a06d1e981933 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 17:58:59 +0000 Subject: [PATCH 1/5] feat(orchestrator): Freeze queue processing while FreezeProcessing is set --- service/submitqueue/README.md | 2 +- .../submitqueue/gateway/server/BUILD.bazel | 4 +- service/submitqueue/gateway/server/main.go | 22 +---- .../orchestrator/server/BUILD.bazel | 10 ++ .../submitqueue/orchestrator/server/main.go | 8 +- .../orchestrator/server/queuepausegate.go | 64 +++++++++++++ .../server/queuepausegate_test.go | 93 +++++++++++++++++++ .../submitqueue/queuepausestore/BUILD.bazel | 26 ++++++ .../queuepausestore/queuepausestore.go | 46 +++++++++ .../queuepausestore/queuepausestore_test.go | 59 ++++++++++++ 10 files changed, 309 insertions(+), 25 deletions(-) create mode 100644 service/submitqueue/orchestrator/server/queuepausegate.go create mode 100644 service/submitqueue/orchestrator/server/queuepausegate_test.go create mode 100644 service/submitqueue/queuepausestore/BUILD.bazel create mode 100644 service/submitqueue/queuepausestore/queuepausestore.go create mode 100644 service/submitqueue/queuepausestore/queuepausestore_test.go diff --git a/service/submitqueue/README.md b/service/submitqueue/README.md index e35933c5a..ff6abb248 100644 --- a/service/submitqueue/README.md +++ b/service/submitqueue/README.md @@ -21,7 +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 | +| `QUEUE_PAUSE_CONFIG_PATH` | both | no | YAML file of per-queue pause switches, re-read on every land and every gated delivery | 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 ba0697afc..275cbaf9a 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -30,11 +30,9 @@ go_library( "//platform/extension/messagequeue/mysql:go_default_library", "//service/messagequeue:go_default_library", "//service/submitqueue/gateway/server/mapper:go_default_library", + "//service/submitqueue/queuepausestore: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 26c4a9e6a..85342e588 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -42,11 +42,9 @@ import ( queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" servicemq "github.com/uber/submitqueue/service/messagequeue" "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper" + "github.com/uber/submitqueue/service/submitqueue/queuepausestore" "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" @@ -171,22 +169,6 @@ func gatewayStatusError(err error) error { } } -// 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 { @@ -287,7 +269,7 @@ func run() error { if err != nil { return fmt.Errorf("failed to load queue configs: %w", err) } - queuePauses, err := newQueuePauseStore(logger) + queuePauses, err := queuepausestore.FromEnv(logger) if err != nil { return err } diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index c5ea7e27d..39ef95aca 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -12,6 +12,7 @@ go_library( "config.go", "main.go", "profiles.go", + "queuepausegate.go", ], importpath = "github.com/uber/submitqueue/service/submitqueue/orchestrator/server", visibility = ["//visibility:private"], @@ -37,6 +38,7 @@ go_library( "//platform/http:go_default_library", "//platform/pipeline:go_default_library", "//service/messagequeue:go_default_library", + "//service/submitqueue/queuepausestore:go_default_library", "//submitqueue/core/changeset:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", @@ -54,6 +56,7 @@ go_library( "//submitqueue/extension/conflict/fake:go_default_library", "//submitqueue/extension/conflict/none:go_default_library", "//submitqueue/extension/conflict/pathoverlap:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", "//submitqueue/extension/speculation/allocator/sticky:go_default_library", "//submitqueue/extension/speculation/generator/bestfirst:go_default_library", "//submitqueue/extension/speculation/scorer:go_default_library", @@ -116,6 +119,7 @@ go_test( "config_test.go", "main_test.go", "profiles_test.go", + "queuepausegate_test.go", ], # The pinned git, for the tests that resolve a git change provider against a # real repository. @@ -126,7 +130,10 @@ go_test( }, deps = [ "//platform/base/change:go_default_library", + "//platform/base/messagequeue:go_default_library", "//platform/errs:go_default_library", + "//platform/extension/consumergate:go_default_library", + "//platform/extension/consumergate/mock:go_default_library", "//platform/git/exectest:go_default_library", "//platform/git/repo:go_default_library", "//platform/githubactions:go_default_library", @@ -134,12 +141,15 @@ go_test( "//submitqueue/extension/buildrunner:go_default_library", "//submitqueue/extension/changeprovider:go_default_library", "//submitqueue/extension/conflict:go_default_library", + "//submitqueue/extension/queuepause/mock:go_default_library", "//submitqueue/extension/speculation/scorer:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", "//submitqueue/orchestrator/extension/storage:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", "@org_uber_go_zap//zaptest:go_default_library", ], ) diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 2f1038c8e..28e54c1c6 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -48,6 +48,7 @@ import ( queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/platform/pipeline" servicemq "github.com/uber/submitqueue/service/messagequeue" + "github.com/uber/submitqueue/service/submitqueue/queuepausestore" "github.com/uber/submitqueue/submitqueue/core/changeset" "github.com/uber/submitqueue/submitqueue/extension/validator" validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake" @@ -223,6 +224,11 @@ func run() error { Hooks: hookResolver{}, } + queuePauses, err := queuepausestore.FromEnv(logger) + if err != nil { + return err + } + // Assemble the pipeline: one call builds the topic registry, creates // primary and DLQ consumers, eagerly constructs all controllers, and // returns a single lifecycle.Component the host drives with Start/Stop. @@ -235,7 +241,7 @@ func run() error { orchestrator.Stages, pipeline.PublishOnly(orchestrator.PublishOnlyTopics...), pipeline.Classifiers(primaryErrorClassifiers()...), - pipeline.Gate(newConsumerGate(logger)), + pipeline.Gate(newQueuePauseGate(logger.Sugar(), queuePauses, newConsumerGate(logger))), ) if err != nil { return fmt.Errorf("failed to construct pipeline: %w", err) diff --git a/service/submitqueue/orchestrator/server/queuepausegate.go b/service/submitqueue/orchestrator/server/queuepausegate.go new file mode 100644 index 000000000..16338e9c6 --- /dev/null +++ b/service/submitqueue/orchestrator/server/queuepausegate.go @@ -0,0 +1,64 @@ +// Copyright (c) 2026 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 main + +import ( + "context" + + "github.com/uber/submitqueue/platform/extension/consumergate" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" + "go.uber.org/zap" +) + +var ( + _ consumergate.Gate = queuePauseGate{} + _ consumergate.Entry = frozenEntry{} +) + +// queuePauseGate closes the gate for a queue while its pause state freezes +// processing, and otherwise defers to the gate it wraps. +type queuePauseGate struct { + logger *zap.SugaredLogger + pauses queuepause.Store + next consumergate.Gate +} + +func newQueuePauseGate(logger *zap.SugaredLogger, pauses queuepause.Store, next consumergate.Gate) queuePauseGate { + return queuePauseGate{logger: logger, pauses: pauses, next: next} +} + +// Enter reads the pause state of the delivery's queue, which is its tenant. A +// failed read freezes the queue: the consumer treats a gate error as open, so +// returning it would unfreeze a queue whenever the pause backend is down. +func (g queuePauseGate) Enter(ctx context.Context, key consumergate.Key) (consumergate.Entry, error) { + state, err := g.pauses.Get(ctx, key.Partition.Tenant) + if err != nil { + g.logger.Errorw("failed to read queue pause state, freezing queue", "queue", key.Partition.Tenant, "error", err) + return frozenEntry{}, nil + } + if state.FreezeProcessing { + return frozenEntry{}, nil + } + return g.next.Enter(ctx, key) +} + +// frozenEntry is a blocked entry that keeps no parked record. +type frozenEntry struct{} + +func (frozenEntry) Blocked() bool { return true } + +func (frozenEntry) Park(context.Context, consumergate.DeliveryDescriptor) error { return nil } + +func (frozenEntry) Unpark(context.Context, consumergate.DeliveryDescriptor) error { return nil } diff --git a/service/submitqueue/orchestrator/server/queuepausegate_test.go b/service/submitqueue/orchestrator/server/queuepausegate_test.go new file mode 100644 index 000000000..0208e15f1 --- /dev/null +++ b/service/submitqueue/orchestrator/server/queuepausegate_test.go @@ -0,0 +1,93 @@ +// Copyright (c) 2026 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 main + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/extension/consumergate" + consumergatemock "github.com/uber/submitqueue/platform/extension/consumergate/mock" + "github.com/uber/submitqueue/submitqueue/entity" + queuepausemock "github.com/uber/submitqueue/submitqueue/extension/queuepause/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +func TestQueuePauseGate_Enter(t *testing.T) { + key := consumergate.Key{ + ConsumerGroup: "orchestrator", + Partition: entityqueue.PartitionIdentity{Tenant: "my-queue", PartitionKey: "p1"}, + } + tests := []struct { + name string + state entity.QueuePauseState + storeErr error + nextBlocked bool + wantBlocked bool + wantNext bool + }{ + {name: "freeze closes the gate", state: entity.QueuePauseState{FreezeProcessing: true}, wantBlocked: true}, + {name: "freeze with reject closes the gate", state: entity.QueuePauseState{FreezeProcessing: true, RejectLands: true}, wantBlocked: true}, + {name: "failed read closes the gate", storeErr: errors.New("pause backend down"), wantBlocked: true}, + {name: "not paused defers to the wrapped gate", state: entity.QueuePauseState{}, wantNext: true}, + {name: "reject only does not freeze", state: entity.QueuePauseState{RejectLands: true}, wantNext: true}, + {name: "wrapped gate can still close it", state: entity.QueuePauseState{}, nextBlocked: true, wantNext: true, wantBlocked: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + pauses := queuepausemock.NewMockStore(ctrl) + pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(tt.state, tt.storeErr) + next := consumergatemock.NewMockGate(ctrl) + if tt.wantNext { + nextEntry := consumergatemock.NewMockEntry(ctrl) + nextEntry.EXPECT().Blocked().Return(tt.nextBlocked).AnyTimes() + next.EXPECT().Enter(gomock.Any(), key).Return(nextEntry, nil) + } + gate := newQueuePauseGate(zap.NewNop().Sugar(), pauses, next) + + entry, err := gate.Enter(context.Background(), key) + + require.NoError(t, err) + assert.Equal(t, tt.wantBlocked, entry.Blocked()) + }) + } +} + +func TestQueuePauseGate_EnterPropagatesWrappedGateError(t *testing.T) { + ctrl := gomock.NewController(t) + key := consumergate.Key{Partition: entityqueue.PartitionIdentity{Tenant: "my-queue"}} + pauses := queuepausemock.NewMockStore(ctrl) + pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePauseState{}, nil) + next := consumergatemock.NewMockGate(ctrl) + next.EXPECT().Enter(gomock.Any(), key).Return(nil, errors.New("gate files unreadable")) + gate := newQueuePauseGate(zap.NewNop().Sugar(), pauses, next) + + _, err := gate.Enter(context.Background(), key) + + require.Error(t, err) +} + +func TestFrozenEntry_ParkAndUnparkKeepNothing(t *testing.T) { + entry := frozenEntry{} + + require.NoError(t, entry.Park(context.Background(), consumergate.DeliveryDescriptor{})) + require.NoError(t, entry.Unpark(context.Background(), consumergate.DeliveryDescriptor{})) +} diff --git a/service/submitqueue/queuepausestore/BUILD.bazel b/service/submitqueue/queuepausestore/BUILD.bazel new file mode 100644 index 000000000..0e1ac6494 --- /dev/null +++ b/service/submitqueue/queuepausestore/BUILD.bazel @@ -0,0 +1,26 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["queuepausestore.go"], + importpath = "github.com/uber/submitqueue/service/submitqueue/queuepausestore", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/extension/queuepause:go_default_library", + "//submitqueue/extension/queuepause/file:go_default_library", + "//submitqueue/extension/queuepause/noop:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["queuepausestore_test.go"], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/service/submitqueue/queuepausestore/queuepausestore.go b/service/submitqueue/queuepausestore/queuepausestore.go new file mode 100644 index 000000000..121df48a5 --- /dev/null +++ b/service/submitqueue/queuepausestore/queuepausestore.go @@ -0,0 +1,46 @@ +// Copyright (c) 2026 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 queuepausestore selects the queuepause.Store a SubmitQueue service +// runs with, so the gateway and orchestrator resolve it the same way. +package queuepausestore + +import ( + "fmt" + "os" + + "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" + "go.uber.org/zap" +) + +const pathEnv = "QUEUE_PAUSE_CONFIG_PATH" + +// FromEnv returns the file-backed store when QUEUE_PAUSE_CONFIG_PATH is set, +// and otherwise a store under which no queue is ever paused. A path that +// cannot be loaded is an error, so a misconfigured service fails at startup. +func FromEnv(logger *zap.Logger) (queuepause.Store, error) { + path := os.Getenv(pathEnv) + 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 +} diff --git a/service/submitqueue/queuepausestore/queuepausestore_test.go b/service/submitqueue/queuepausestore/queuepausestore_test.go new file mode 100644 index 000000000..ff72016b4 --- /dev/null +++ b/service/submitqueue/queuepausestore/queuepausestore_test.go @@ -0,0 +1,59 @@ +// Copyright (c) 2026 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 queuepausestore + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" + "go.uber.org/zap" +) + +func TestFromEnv(t *testing.T) { + pausedFile := filepath.Join(t.TempDir(), "pause.yaml") + require.NoError(t, os.WriteFile(pausedFile, []byte("queues:\n - name: q\n freeze_processing: true\n"), 0o600)) + + tests := []struct { + name string + path string + wantErr bool + wantState entity.QueuePauseState + }{ + {name: "unset pauses nothing", path: "", wantState: entity.QueuePauseState{}}, + {name: "file store reads the configured file", path: pausedFile, wantState: entity.QueuePauseState{FreezeProcessing: true}}, + {name: "unreadable path fails", path: filepath.Join(t.TempDir(), "absent.yaml"), wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv(pathEnv, tt.path) + + store, err := FromEnv(zap.NewNop()) + + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + got, err := store.Get(context.Background(), "q") + require.NoError(t, err) + assert.Equal(t, tt.wantState, got) + }) + } +} From ae0e4bea5615cacc40433acb39e5a79f5ede1d8c Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 18:43:43 +0000 Subject: [PATCH 2/5] fix(orchestrator): Pass a cancelled context through the queue pause gate --- .../orchestrator/server/queuepausegate.go | 8 +++++++- .../orchestrator/server/queuepausegate_test.go | 13 +++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/service/submitqueue/orchestrator/server/queuepausegate.go b/service/submitqueue/orchestrator/server/queuepausegate.go index 16338e9c6..7431f28cc 100644 --- a/service/submitqueue/orchestrator/server/queuepausegate.go +++ b/service/submitqueue/orchestrator/server/queuepausegate.go @@ -16,6 +16,7 @@ package main import ( "context" + "errors" "github.com/uber/submitqueue/platform/extension/consumergate" "github.com/uber/submitqueue/submitqueue/extension/queuepause" @@ -41,9 +42,14 @@ func newQueuePauseGate(logger *zap.SugaredLogger, pauses queuepause.Store, next // Enter reads the pause state of the delivery's queue, which is its tenant. A // failed read freezes the queue: the consumer treats a gate error as open, so -// returning it would unfreeze a queue whenever the pause backend is down. +// returning it would unfreeze a queue whenever the pause backend is down. A +// cancelled context is the exception: the consumer is shutting down and +// handles that error itself. func (g queuePauseGate) Enter(ctx context.Context, key consumergate.Key) (consumergate.Entry, error) { state, err := g.pauses.Get(ctx, key.Partition.Tenant) + if errors.Is(err, context.Canceled) { + return nil, err + } if err != nil { g.logger.Errorw("failed to read queue pause state, freezing queue", "queue", key.Partition.Tenant, "error", err) return frozenEntry{}, nil diff --git a/service/submitqueue/orchestrator/server/queuepausegate_test.go b/service/submitqueue/orchestrator/server/queuepausegate_test.go index 0208e15f1..3c383be0f 100644 --- a/service/submitqueue/orchestrator/server/queuepausegate_test.go +++ b/service/submitqueue/orchestrator/server/queuepausegate_test.go @@ -17,6 +17,7 @@ package main import ( "context" "errors" + "fmt" "testing" "github.com/stretchr/testify/assert" @@ -85,6 +86,18 @@ func TestQueuePauseGate_EnterPropagatesWrappedGateError(t *testing.T) { require.Error(t, err) } +func TestQueuePauseGate_EnterReturnsCancelledContextError(t *testing.T) { + ctrl := gomock.NewController(t) + key := consumergate.Key{Partition: entityqueue.PartitionIdentity{Tenant: "my-queue"}} + pauses := queuepausemock.NewMockStore(ctrl) + pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePauseState{}, fmt.Errorf("read pause state: %w", context.Canceled)) + gate := newQueuePauseGate(zap.NewNop().Sugar(), pauses, consumergatemock.NewMockGate(ctrl)) + + _, err := gate.Enter(context.Background(), key) + + require.ErrorIs(t, err, context.Canceled) +} + func TestFrozenEntry_ParkAndUnparkKeepNothing(t *testing.T) { entry := frozenEntry{} From 5bc69faa0c70f5921f0e1cc12e3a8a892a780fc5 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:34:26 +0000 Subject: [PATCH 3/5] refactor(orchestrator): Use the renamed QueuePause entity in pause tests --- .../orchestrator/server/queuepausegate_test.go | 16 ++++++++-------- .../queuepausestore/queuepausestore_test.go | 6 +++--- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/service/submitqueue/orchestrator/server/queuepausegate_test.go b/service/submitqueue/orchestrator/server/queuepausegate_test.go index 3c383be0f..883a758b8 100644 --- a/service/submitqueue/orchestrator/server/queuepausegate_test.go +++ b/service/submitqueue/orchestrator/server/queuepausegate_test.go @@ -38,18 +38,18 @@ func TestQueuePauseGate_Enter(t *testing.T) { } tests := []struct { name string - state entity.QueuePauseState + state entity.QueuePause storeErr error nextBlocked bool wantBlocked bool wantNext bool }{ - {name: "freeze closes the gate", state: entity.QueuePauseState{FreezeProcessing: true}, wantBlocked: true}, - {name: "freeze with reject closes the gate", state: entity.QueuePauseState{FreezeProcessing: true, RejectLands: true}, wantBlocked: true}, + {name: "freeze closes the gate", state: entity.QueuePause{FreezeProcessing: true}, wantBlocked: true}, + {name: "freeze with reject closes the gate", state: entity.QueuePause{FreezeProcessing: true, RejectLands: true}, wantBlocked: true}, {name: "failed read closes the gate", storeErr: errors.New("pause backend down"), wantBlocked: true}, - {name: "not paused defers to the wrapped gate", state: entity.QueuePauseState{}, wantNext: true}, - {name: "reject only does not freeze", state: entity.QueuePauseState{RejectLands: true}, wantNext: true}, - {name: "wrapped gate can still close it", state: entity.QueuePauseState{}, nextBlocked: true, wantNext: true, wantBlocked: true}, + {name: "not paused defers to the wrapped gate", state: entity.QueuePause{}, wantNext: true}, + {name: "reject only does not freeze", state: entity.QueuePause{RejectLands: true}, wantNext: true}, + {name: "wrapped gate can still close it", state: entity.QueuePause{}, nextBlocked: true, wantNext: true, wantBlocked: true}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -76,7 +76,7 @@ func TestQueuePauseGate_EnterPropagatesWrappedGateError(t *testing.T) { ctrl := gomock.NewController(t) key := consumergate.Key{Partition: entityqueue.PartitionIdentity{Tenant: "my-queue"}} pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePauseState{}, nil) + pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePause{}, nil) next := consumergatemock.NewMockGate(ctrl) next.EXPECT().Enter(gomock.Any(), key).Return(nil, errors.New("gate files unreadable")) gate := newQueuePauseGate(zap.NewNop().Sugar(), pauses, next) @@ -90,7 +90,7 @@ func TestQueuePauseGate_EnterReturnsCancelledContextError(t *testing.T) { ctrl := gomock.NewController(t) key := consumergate.Key{Partition: entityqueue.PartitionIdentity{Tenant: "my-queue"}} pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePauseState{}, fmt.Errorf("read pause state: %w", context.Canceled)) + pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePause{}, fmt.Errorf("read pause state: %w", context.Canceled)) gate := newQueuePauseGate(zap.NewNop().Sugar(), pauses, consumergatemock.NewMockGate(ctrl)) _, err := gate.Enter(context.Background(), key) diff --git a/service/submitqueue/queuepausestore/queuepausestore_test.go b/service/submitqueue/queuepausestore/queuepausestore_test.go index ff72016b4..1dd120737 100644 --- a/service/submitqueue/queuepausestore/queuepausestore_test.go +++ b/service/submitqueue/queuepausestore/queuepausestore_test.go @@ -34,10 +34,10 @@ func TestFromEnv(t *testing.T) { name string path string wantErr bool - wantState entity.QueuePauseState + wantState entity.QueuePause }{ - {name: "unset pauses nothing", path: "", wantState: entity.QueuePauseState{}}, - {name: "file store reads the configured file", path: pausedFile, wantState: entity.QueuePauseState{FreezeProcessing: true}}, + {name: "unset pauses nothing", path: "", wantState: entity.QueuePause{}}, + {name: "file store reads the configured file", path: pausedFile, wantState: entity.QueuePause{FreezeProcessing: true}}, {name: "unreadable path fails", path: filepath.Join(t.TempDir(), "absent.yaml"), wantErr: true}, } for _, tt := range tests { From a2a1d602630399adc8081313fe47dc9ab0c7f846 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:37:20 +0000 Subject: [PATCH 4/5] test(orchestrator): Simulate queue pause state with the fake store --- service/submitqueue/orchestrator/server/BUILD.bazel | 2 +- .../orchestrator/server/queuepausegate_test.go | 11 ++++------- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index 39ef95aca..6b422a2e4 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -141,7 +141,7 @@ go_test( "//submitqueue/extension/buildrunner:go_default_library", "//submitqueue/extension/changeprovider:go_default_library", "//submitqueue/extension/conflict:go_default_library", - "//submitqueue/extension/queuepause/mock:go_default_library", + "//submitqueue/extension/queuepause/fake:go_default_library", "//submitqueue/extension/speculation/scorer:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", "//submitqueue/orchestrator/extension/storage:go_default_library", diff --git a/service/submitqueue/orchestrator/server/queuepausegate_test.go b/service/submitqueue/orchestrator/server/queuepausegate_test.go index 883a758b8..31989abef 100644 --- a/service/submitqueue/orchestrator/server/queuepausegate_test.go +++ b/service/submitqueue/orchestrator/server/queuepausegate_test.go @@ -26,7 +26,7 @@ import ( "github.com/uber/submitqueue/platform/extension/consumergate" consumergatemock "github.com/uber/submitqueue/platform/extension/consumergate/mock" "github.com/uber/submitqueue/submitqueue/entity" - queuepausemock "github.com/uber/submitqueue/submitqueue/extension/queuepause/mock" + queuepausefake "github.com/uber/submitqueue/submitqueue/extension/queuepause/fake" "go.uber.org/mock/gomock" "go.uber.org/zap" ) @@ -54,8 +54,7 @@ func TestQueuePauseGate_Enter(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { ctrl := gomock.NewController(t) - pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(tt.state, tt.storeErr) + pauses := queuepausefake.New().Set("my-queue", tt.state).FailWith(tt.storeErr) next := consumergatemock.NewMockGate(ctrl) if tt.wantNext { nextEntry := consumergatemock.NewMockEntry(ctrl) @@ -75,8 +74,7 @@ func TestQueuePauseGate_Enter(t *testing.T) { func TestQueuePauseGate_EnterPropagatesWrappedGateError(t *testing.T) { ctrl := gomock.NewController(t) key := consumergate.Key{Partition: entityqueue.PartitionIdentity{Tenant: "my-queue"}} - pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePause{}, nil) + pauses := queuepausefake.New() next := consumergatemock.NewMockGate(ctrl) next.EXPECT().Enter(gomock.Any(), key).Return(nil, errors.New("gate files unreadable")) gate := newQueuePauseGate(zap.NewNop().Sugar(), pauses, next) @@ -89,8 +87,7 @@ func TestQueuePauseGate_EnterPropagatesWrappedGateError(t *testing.T) { func TestQueuePauseGate_EnterReturnsCancelledContextError(t *testing.T) { ctrl := gomock.NewController(t) key := consumergate.Key{Partition: entityqueue.PartitionIdentity{Tenant: "my-queue"}} - pauses := queuepausemock.NewMockStore(ctrl) - pauses.EXPECT().Get(gomock.Any(), "my-queue").Return(entity.QueuePause{}, fmt.Errorf("read pause state: %w", context.Canceled)) + pauses := queuepausefake.New().FailWith(fmt.Errorf("read pause state: %w", context.Canceled)) gate := newQueuePauseGate(zap.NewNop().Sugar(), pauses, consumergatemock.NewMockGate(ctrl)) _, err := gate.Enter(context.Background(), key) From 2459caa028a86878d497a88384211c063d1bae8a Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:41:11 +0000 Subject: [PATCH 5/5] refactor(service): Rename queuepausestore to core --- .../submitqueue/{queuepausestore => core}/BUILD.bazel | 6 +++--- .../queuepausestore.go => core/queuepause.go} | 10 +++++----- .../queuepause_test.go} | 6 +++--- service/submitqueue/gateway/server/BUILD.bazel | 2 +- service/submitqueue/gateway/server/main.go | 4 ++-- service/submitqueue/orchestrator/server/BUILD.bazel | 2 +- service/submitqueue/orchestrator/server/main.go | 4 ++-- 7 files changed, 17 insertions(+), 17 deletions(-) rename service/submitqueue/{queuepausestore => core}/BUILD.bazel (89%) rename service/submitqueue/{queuepausestore/queuepausestore.go => core/queuepause.go} (81%) rename service/submitqueue/{queuepausestore/queuepausestore_test.go => core/queuepause_test.go} (93%) diff --git a/service/submitqueue/queuepausestore/BUILD.bazel b/service/submitqueue/core/BUILD.bazel similarity index 89% rename from service/submitqueue/queuepausestore/BUILD.bazel rename to service/submitqueue/core/BUILD.bazel index 0e1ac6494..74d63f0fa 100644 --- a/service/submitqueue/queuepausestore/BUILD.bazel +++ b/service/submitqueue/core/BUILD.bazel @@ -2,8 +2,8 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["queuepausestore.go"], - importpath = "github.com/uber/submitqueue/service/submitqueue/queuepausestore", + srcs = ["queuepause.go"], + importpath = "github.com/uber/submitqueue/service/submitqueue/core", visibility = ["//visibility:public"], deps = [ "//submitqueue/extension/queuepause:go_default_library", @@ -15,7 +15,7 @@ go_library( go_test( name = "go_default_test", - srcs = ["queuepausestore_test.go"], + srcs = ["queuepause_test.go"], embed = [":go_default_library"], deps = [ "//submitqueue/entity:go_default_library", diff --git a/service/submitqueue/queuepausestore/queuepausestore.go b/service/submitqueue/core/queuepause.go similarity index 81% rename from service/submitqueue/queuepausestore/queuepausestore.go rename to service/submitqueue/core/queuepause.go index 121df48a5..db601651d 100644 --- a/service/submitqueue/queuepausestore/queuepausestore.go +++ b/service/submitqueue/core/queuepause.go @@ -12,9 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package queuepausestore selects the queuepause.Store a SubmitQueue service -// runs with, so the gateway and orchestrator resolve it the same way. -package queuepausestore +// Package core holds wiring that both SubmitQueue services resolve the same +// way, so the gateway and orchestrator share one implementation. +package core import ( "fmt" @@ -28,10 +28,10 @@ import ( const pathEnv = "QUEUE_PAUSE_CONFIG_PATH" -// FromEnv returns the file-backed store when QUEUE_PAUSE_CONFIG_PATH is set, +// QueuePauseStoreFromEnv returns the file-backed store when QUEUE_PAUSE_CONFIG_PATH is set, // and otherwise a store under which no queue is ever paused. A path that // cannot be loaded is an error, so a misconfigured service fails at startup. -func FromEnv(logger *zap.Logger) (queuepause.Store, error) { +func QueuePauseStoreFromEnv(logger *zap.Logger) (queuepause.Store, error) { path := os.Getenv(pathEnv) if path == "" { logger.Info("queue pause disabled") diff --git a/service/submitqueue/queuepausestore/queuepausestore_test.go b/service/submitqueue/core/queuepause_test.go similarity index 93% rename from service/submitqueue/queuepausestore/queuepausestore_test.go rename to service/submitqueue/core/queuepause_test.go index 1dd120737..4fc1a969a 100644 --- a/service/submitqueue/queuepausestore/queuepausestore_test.go +++ b/service/submitqueue/core/queuepause_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package queuepausestore +package core import ( "context" @@ -26,7 +26,7 @@ import ( "go.uber.org/zap" ) -func TestFromEnv(t *testing.T) { +func TestQueuePauseStoreFromEnv(t *testing.T) { pausedFile := filepath.Join(t.TempDir(), "pause.yaml") require.NoError(t, os.WriteFile(pausedFile, []byte("queues:\n - name: q\n freeze_processing: true\n"), 0o600)) @@ -44,7 +44,7 @@ func TestFromEnv(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Setenv(pathEnv, tt.path) - store, err := FromEnv(zap.NewNop()) + store, err := QueuePauseStoreFromEnv(zap.NewNop()) if tt.wantErr { require.Error(t, err) diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 275cbaf9a..0a2d9c205 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -29,8 +29,8 @@ go_library( "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", "//service/messagequeue:go_default_library", + "//service/submitqueue/core:go_default_library", "//service/submitqueue/gateway/server/mapper:go_default_library", - "//service/submitqueue/queuepausestore:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/extension/queueconfig/yaml: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 85342e588..91cfcdd05 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -41,8 +41,8 @@ import ( extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" servicemq "github.com/uber/submitqueue/service/messagequeue" + "github.com/uber/submitqueue/service/submitqueue/core" "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper" - "github.com/uber/submitqueue/service/submitqueue/queuepausestore" "github.com/uber/submitqueue/submitqueue/core/topickey" yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml" "github.com/uber/submitqueue/submitqueue/gateway/controller" @@ -269,7 +269,7 @@ func run() error { if err != nil { return fmt.Errorf("failed to load queue configs: %w", err) } - queuePauses, err := queuepausestore.FromEnv(logger) + queuePauses, err := core.QueuePauseStoreFromEnv(logger) if err != nil { return err } diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index 6b422a2e4..751d844e7 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -38,7 +38,7 @@ go_library( "//platform/http:go_default_library", "//platform/pipeline:go_default_library", "//service/messagequeue:go_default_library", - "//service/submitqueue/queuepausestore:go_default_library", + "//service/submitqueue/core:go_default_library", "//submitqueue/core/changeset:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 28e54c1c6..8203de747 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -48,7 +48,7 @@ import ( queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/platform/pipeline" servicemq "github.com/uber/submitqueue/service/messagequeue" - "github.com/uber/submitqueue/service/submitqueue/queuepausestore" + "github.com/uber/submitqueue/service/submitqueue/core" "github.com/uber/submitqueue/submitqueue/core/changeset" "github.com/uber/submitqueue/submitqueue/extension/validator" validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake" @@ -224,7 +224,7 @@ func run() error { Hooks: hookResolver{}, } - queuePauses, err := queuepausestore.FromEnv(logger) + queuePauses, err := core.QueuePauseStoreFromEnv(logger) if err != nil { return err }