From 837041dd31c50f4faaa2b9758ff9a06d1e981933 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 17:58:59 +0000 Subject: [PATCH 1/7] 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/7] 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/7] 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/7] 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/7] 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 } From abad345cacae82c210c28bd5427c47a70bf6a37d Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 19:12:15 +0000 Subject: [PATCH 6/7] test(e2e): Cover queue pause and freeze across every switch combination --- service/submitqueue/docker-compose.yml | 16 +- .../gateway/server/docker-compose.yml | 2 +- .../submitqueue/gateway/server/queues.yaml | 3 + test/e2e/submitqueue/BUILD.bazel | 1 + test/e2e/submitqueue/queuepause_test.go | 158 ++++++++++++++++++ test/e2e/submitqueue/suite_test.go | 11 ++ 6 files changed, 187 insertions(+), 4 deletions(-) create mode 100644 test/e2e/submitqueue/queuepause_test.go diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml index ef1a094e4..1a566b4f2 100644 --- a/service/submitqueue/docker-compose.yml +++ b/service/submitqueue/docker-compose.yml @@ -13,6 +13,12 @@ # individual queue controllers by writing/removing gate files from the host. # Override the host side with SQ_CONSUMER_GATE_DIR (the e2e suite points it at # a per-run temp dir); it defaults to /tmp/sq-consumergate for local runs. +# Queue pause: the gateway and orchestrator read a pause file from a shared host +# directory (bind-mounted at /var/submitqueue/queuepause), so E2E tests can pause +# and unpause queues by rewriting it. Override the host side with +# SQ_QUEUE_PAUSE_DIR and set SQ_QUEUE_PAUSE_CONFIG_PATH to the file's path inside +# the container. It is unset by default, which leaves pausing disabled, because +# the file store requires its file to exist when the service starts. # SQ_CONTAINER_USER controls the UID:GID of the application services that write # parked records into that bind mount. It defaults to root for normal local # usage; the e2e suite selects the host user for rootful Docker and root for @@ -77,15 +83,17 @@ services: # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue} + - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue,e2e-pause-queue} # Path to YAML queue configuration baked into the image - QUEUE_CONFIG_PATH=/app/queues.yaml # Stable subscriber name for the request-log consumer - HOSTNAME=gateway-dev # Consumer-gate state shared with the host (see header comment) - CONSUMER_GATE_DIR=/var/submitqueue/consumergate + - QUEUE_PAUSE_CONFIG_PATH=${SQ_QUEUE_PAUSE_CONFIG_PATH:-} volumes: - ${SQ_CONSUMER_GATE_DIR:-/tmp/sq-consumergate}:/var/submitqueue/consumergate + - ${SQ_QUEUE_PAUSE_DIR:-/tmp/sq-queuepause}:/var/submitqueue/queuepause depends_on: mysql-app: condition: service_healthy @@ -110,12 +118,14 @@ services: # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue} + - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue,e2e-pause-queue} - HOSTNAME=orchestrator-dev # Consumer-gate state shared with the host (see header comment) - CONSUMER_GATE_DIR=/var/submitqueue/consumergate + - QUEUE_PAUSE_CONFIG_PATH=${SQ_QUEUE_PAUSE_CONFIG_PATH:-} volumes: - ${SQ_CONSUMER_GATE_DIR:-/tmp/sq-consumergate}:/var/submitqueue/consumergate + - ${SQ_QUEUE_PAUSE_DIR:-/tmp/sq-queuepause}:/var/submitqueue/queuepause depends_on: mysql-app: condition: service_healthy @@ -143,7 +153,7 @@ services: # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue} + - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue,e2e-pause-queue} - HOSTNAME=runway-dev # Consumer-gate state shared with the host (see header comment) - CONSUMER_GATE_DIR=/var/submitqueue/consumergate diff --git a/service/submitqueue/gateway/server/docker-compose.yml b/service/submitqueue/gateway/server/docker-compose.yml index d44b1233b..cdf388104 100644 --- a/service/submitqueue/gateway/server/docker-compose.yml +++ b/service/submitqueue/gateway/server/docker-compose.yml @@ -66,7 +66,7 @@ services: # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue} + - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,e2e-github-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue,e2e-pause-queue} # Path to YAML queue configuration baked into the image - QUEUE_CONFIG_PATH=/app/queues.yaml # Stable subscriber name for the request-log consumer diff --git a/service/submitqueue/gateway/server/queues.yaml b/service/submitqueue/gateway/server/queues.yaml index 8a292266b..839021568 100644 --- a/service/submitqueue/gateway/server/queues.yaml +++ b/service/submitqueue/gateway/server/queues.yaml @@ -39,3 +39,6 @@ queues: # Exercises path-overlap conflict analysis in the built-in orchestrator # profile. - name: file-overlap-queue + # Owns the pause and unpause e2e tests, which switch its pause state. It is + # a baseline-profile queue, so pausing it never touches another test's queue. + - name: e2e-pause-queue diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel index d56e86138..9c76a683d 100644 --- a/test/e2e/submitqueue/BUILD.bazel +++ b/test/e2e/submitqueue/BUILD.bazel @@ -7,6 +7,7 @@ go_test( "git_suite_test.go", "github_suite_test.go", "harness_test.go", + "queuepause_test.go", "suite_test.go", ], data = [ diff --git a/test/e2e/submitqueue/queuepause_test.go b/test/e2e/submitqueue/queuepause_test.go new file mode 100644 index 000000000..f2beea221 --- /dev/null +++ b/test/e2e/submitqueue/queuepause_test.go @@ -0,0 +1,158 @@ +// 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 e2e_test + +import ( + "fmt" + "os" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + changepb "github.com/uber/submitqueue/api/base/change/protopb" + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/submitqueue/entity" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const ( + queuePauseFileName = "pause.yaml" + + // pauseQueue is owned by these tests, so switching its pause state cannot + // disturb another test's queue. + pauseQueue = "e2e-pause-queue" + + // sentinelQueue is never paused. Its request landing proves the + // orchestrator is running every stage, so a frozen queue that has not + // advanced was held and not merely slow. + sentinelQueue = "e2e-test-queue" +) + +// writeQueuePauseFile replaces the pause file by rename inside its directory. +// The services read the file while it is rewritten, so a reader must see either +// the old or the new contents and never a partial file. The directory is +// bind-mounted, so a renamed file is visible in the containers; a single +// bind-mounted file would keep the old inode. +func (s *E2EIntegrationSuite) writeQueuePauseFile(contents string) { + t := s.T() + tmp := s.pausePath + ".tmp" + require.NoError(t, os.WriteFile(tmp, []byte(contents), 0o644), "failed to write pause file") + require.NoError(t, os.Rename(tmp, s.pausePath), "failed to replace pause file") +} + +// setQueuePause sets the pause state of pauseQueue, including its reason, and +// leaves every other queue unpaused. +func (s *E2EIntegrationSuite) setQueuePause(state entity.QueuePause) { + s.writeQueuePauseFile(fmt.Sprintf( + "queues:\n - name: %s\n reject_lands: %t\n freeze_processing: %t\n reason: %q\n", + pauseQueue, state.RejectLands, state.FreezeProcessing, state.Reason)) + s.log.Logf("Set pause state for %s: reject_lands=%t freeze_processing=%t reason=%q", pauseQueue, state.RejectLands, state.FreezeProcessing, state.Reason) +} + +// clearQueuePause unpauses every queue. Tests defer it so a failed assertion +// cannot leave the queue paused for the tests after it. +func (s *E2EIntegrationSuite) clearQueuePause() { + s.writeQueuePauseFile("queues: []\n") +} + +// assertLandRejectedAsPaused lands directly on the gateway and expects the +// paused-queue rejection, which carries no sqid because no request is created. +// A non-empty wantReason must appear in the rejection the caller sees. +func (s *E2EIntegrationSuite) assertLandRejectedAsPaused(queue string, uri string, wantReason string) { + resp, err := s.gatewayClient.Land(s.ctx, &gatewaypb.LandRequest{ + Queue: queue, + Change: &changepb.Change{Uris: []string{uri}}, + Strategy: mergestrategypb.Strategy_REBASE, + }) + require.Error(s.T(), err, "Land on paused queue %s should be rejected", queue) + assert.Equal(s.T(), codes.FailedPrecondition, status.Code(err), "paused queue should map to FailedPrecondition") + assert.Empty(s.T(), resp.GetSqid(), "a rejected land must not return a request") + if wantReason != "" { + assert.Contains(s.T(), status.Convert(err).Message(), wantReason, "rejection should carry the pause reason") + } +} + +// assertHeldWhileSentinelLands proves a frozen request did not advance. The +// sentinel queue shares the orchestrator but not the frozen queue's partitions, +// so once it has landed every stage has run and the frozen request, which was +// published earlier, would have moved if the freeze had not held it. The freeze +// gate records no parked delivery, so this ordering is the only witness. +func (s *E2EIntegrationSuite) assertHeldWhileSentinelLands(held request, sentinelURI string) { + sentinel := s.land(sentinelQueue, sentinelURI) + s.awaitStatus(sentinel, entity.RequestStatusLanded) + assert.Equal(s.T(), entity.RequestStatusAccepted, s.mustStatus(held), + "request %s on a frozen queue must stay accepted while another queue lands", held.sqid) +} + +func uriForPull(repo string, pull int) string { + return fmt.Sprintf("github://github.example.com/uber/%s/pull/%d/abcdef0123456789abcdef0123456789abcdef01", repo, pull) +} + +// Neither switch set: an explicit all-off entry must not pause anything. +func (s *E2EIntegrationSuite) TestQueuePause_Running_LandsAndAdvances() { + s.setQueuePause(entity.QueuePause{}) + defer s.clearQueuePause() + + req := s.land(pauseQueue, uriForPull("e2e-pause", 1001)) + + s.awaitStatus(req, entity.RequestStatusLanded) +} + +// Reject only (drain): in-flight work finishes and new lands are refused. +// The request is held by a freeze first so it is provably in flight when the +// freeze is lifted and rejection begins. +func (s *E2EIntegrationSuite) TestQueuePause_Drain_FinishesInFlightAndRejectsNewLands() { + s.setQueuePause(entity.QueuePause{FreezeProcessing: true}) + defer s.clearQueuePause() + inFlight := s.land(pauseQueue, uriForPull("e2e-pause", 2001)) + s.assertHeldWhileSentinelLands(inFlight, uriForPull("e2e-pause-sentinel", 2002)) + + s.setQueuePause(entity.QueuePause{RejectLands: true}) + + s.assertLandRejectedAsPaused(pauseQueue, uriForPull("e2e-pause", 2003), "") + s.awaitStatus(inFlight, entity.RequestStatusLanded) +} + +// Freeze only (hold): requests are accepted and durable, and nothing advances +// until the freeze is lifted. +func (s *E2EIntegrationSuite) TestQueuePause_Hold_AcceptsButDoesNotAdvance() { + s.setQueuePause(entity.QueuePause{FreezeProcessing: true}) + defer s.clearQueuePause() + held := s.land(pauseQueue, uriForPull("e2e-pause", 3001)) + s.assertHeldWhileSentinelLands(held, uriForPull("e2e-pause-sentinel", 3002)) + + s.clearQueuePause() + + s.awaitStatus(held, entity.RequestStatusLanded) +} + +// Both switches set: new lands are refused with the operator's reason and +// accepted work stays held, and clearing both lets the held request finish. +func (s *E2EIntegrationSuite) TestQueuePause_FullPause_RejectsAndHolds() { + s.setQueuePause(entity.QueuePause{FreezeProcessing: true}) + defer s.clearQueuePause() + held := s.land(pauseQueue, uriForPull("e2e-pause", 4001)) + + const reason = "incident 4242, back at 17:00" + s.setQueuePause(entity.QueuePause{RejectLands: true, FreezeProcessing: true, Reason: reason}) + + s.assertLandRejectedAsPaused(pauseQueue, uriForPull("e2e-pause", 4002), reason) + s.assertHeldWhileSentinelLands(held, uriForPull("e2e-pause-sentinel", 4003)) + + s.clearQueuePause() + + s.awaitStatus(held, entity.RequestStatusLanded) +} diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index 26fa6a48b..44d04c937 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -30,6 +30,7 @@ import ( "fmt" "os" "os/exec" + "path/filepath" "strings" "testing" "time" @@ -61,6 +62,7 @@ type E2EIntegrationSuite struct { queueDB *sql.DB // Queue database appStorage *storagemysql.Storage // White-box view of the operating store (app DB), resolved per queue gate *consumergatefile.Store // Consumer-gate control plane (shared dir bind-mounted into services) + pausePath string // Queue pause file on the host (its directory is bind-mounted into the gateway and orchestrator) } func TestE2EIntegration(t *testing.T) { @@ -95,6 +97,15 @@ func (s *E2EIntegrationSuite) SetupSuite() { t.Setenv("SQ_CONSUMER_GATE_DIR", gateDir) s.gate = consumergatefile.New(gateDir) + // Queue-pause state is an explicit E2E-only opt-in too. The gateway and + // orchestrator read this file on every land and every delivery, and the + // store requires it to exist at startup, so it starts with nothing paused. + pauseDir := t.TempDir() + s.pausePath = filepath.Join(pauseDir, queuePauseFileName) + s.writeQueuePauseFile("queues: []\n") + t.Setenv("SQ_QUEUE_PAUSE_DIR", pauseDir) + t.Setenv("SQ_QUEUE_PAUSE_CONFIG_PATH", "/var/submitqueue/queuepause/"+queuePauseFileName) + // Use docker-compose from service/submitqueue (full stack), resolved from // the test runfiles. All three service images are built from a staged // build context assembled entirely from declared data dependencies. From 665054ce3839eaefab0d5ac9ce86a956cd7b0306 Mon Sep 17 00:00:00 2001 From: manjari Date: Thu, 15 Oct 2026 00:02:23 +0000 Subject: [PATCH 7/7] test(integration): Expect the pause e2e queue in the gateway queue list --- test/integration/submitqueue/gateway/suite_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index da2789f86..cadaa69f9 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -153,6 +153,7 @@ func (s *GatewayIntegrationSuite) TestListQueuesAPI() { "e2e-conflict-error-queue", "e2e-git-queue", "e2e-github-queue", + "e2e-pause-queue", "e2e-redelivery-queue", "e2e-respeculate-queue", "e2e-strand-queue",