test(e2e): Cover queue pause and freeze across every switch combination

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` |

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",
+    ],
+)

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
+}

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)
+		})
+	}
+}

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

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",

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

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
 	}

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

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,20 +130,26 @@ 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",
         "//submitqueue/entity:go_default_library",
         "//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",
     ],
 )

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)