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/core/BUILD.bazel b/service/submitqueue/core/BUILD.bazel new file mode 100644 index 000000000..74d63f0fa --- /dev/null +++ b/service/submitqueue/core/BUILD.bazel @@ -0,0 +1,26 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["queuepause.go"], + importpath = "github.com/uber/submitqueue/service/submitqueue/core", + 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 = ["queuepause_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/core/queuepause.go b/service/submitqueue/core/queuepause.go new file mode 100644 index 000000000..db601651d --- /dev/null +++ b/service/submitqueue/core/queuepause.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 core holds wiring that both SubmitQueue services resolve the same +// way, so the gateway and orchestrator share one implementation. +package core + +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" + +// 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 QueuePauseStoreFromEnv(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/core/queuepause_test.go b/service/submitqueue/core/queuepause_test.go new file mode 100644 index 000000000..4fc1a969a --- /dev/null +++ b/service/submitqueue/core/queuepause_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 core + +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 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)) + + tests := []struct { + name string + path string + wantErr bool + wantState entity.QueuePause + }{ + {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 { + t.Run(tt.name, func(t *testing.T) { + t.Setenv(pathEnv, tt.path) + + store, err := QueuePauseStoreFromEnv(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) + }) + } +} 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/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index ba0697afc..0a2d9c205 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -29,12 +29,10 @@ 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", "//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/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/main.go b/service/submitqueue/gateway/server/main.go index 26c4a9e6a..91cfcdd05 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -41,12 +41,10 @@ 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/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 := core.QueuePauseStoreFromEnv(logger) if err != nil { return err } 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/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index c5ea7e27d..751d844e7 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/core: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/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", "@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..8203de747 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/core" "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 := core.QueuePauseStoreFromEnv(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..7431f28cc --- /dev/null +++ b/service/submitqueue/orchestrator/server/queuepausegate.go @@ -0,0 +1,70 @@ +// 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" + + "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. 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 + } + 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..31989abef --- /dev/null +++ b/service/submitqueue/orchestrator/server/queuepausegate_test.go @@ -0,0 +1,103 @@ +// 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" + "fmt" + "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" + queuepausefake "github.com/uber/submitqueue/submitqueue/extension/queuepause/fake" + "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.QueuePause + storeErr error + nextBlocked bool + wantBlocked bool + wantNext bool + }{ + {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.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) { + ctrl := gomock.NewController(t) + pauses := queuepausefake.New().Set("my-queue", tt.state).FailWith(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 := 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) + + _, err := gate.Enter(context.Background(), key) + + require.Error(t, err) +} + +func TestQueuePauseGate_EnterReturnsCancelledContextError(t *testing.T) { + ctrl := gomock.NewController(t) + key := consumergate.Key{Partition: entityqueue.PartitionIdentity{Tenant: "my-queue"}} + 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) + + require.ErrorIs(t, err, context.Canceled) +} + +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/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. 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",