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