feat(orchestrator): Freeze queue processing while FreezeProcessing is set (#865)

## Why?
A paused queue needs to stop advancing accepted requests without a
redeploy. The consumer gate already lets a controller be stopped and
started at runtime, so freezing a queue is a gate that reads the queue's
pause state.

## What?
- A gate that wraps the orchestrator's existing consumer gate. While
`FreezeProcessing` is set for a delivery's queue (its tenant), the gate
is closed for every stage, including the dead-letter consumers. Blocked
deliveries are postponed without spending retry budget, and released
within a re-check delay of the flag being cleared. Otherwise the wrapped
gate decides, so `CONSUMER_GATE_DIR` keeps working.
- A failed pause read also closes the gate.
- The orchestrator reads `QUEUE_PAUSE_CONFIG_PATH`. The store selection
moves into a shared `queuepausestore` package that the gateway now uses
too.

## Test Plan
`bazel test //service/submitqueue/queuepausestore/...
//service/submitqueue/orchestrator/server`
CI
End to End tests in next PR

## Issue

Author: manjari25

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/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/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/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)

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 }