From c600cc171c32a9b3776f0194f664e879addc2a08 Mon Sep 17 00:00:00 2001 From: manjari Date: Tue, 13 Oct 2026 19:33:23 +0000 Subject: [PATCH 1/4] feat(submitqueue): Add noop and file queuepause store --- submitqueue/extension/queuepause/README.md | 7 + .../extension/queuepause/file/BUILD.bazel | 24 +++ submitqueue/extension/queuepause/file/file.go | 109 ++++++++++++++ .../extension/queuepause/file/file_test.go | 139 ++++++++++++++++++ .../extension/queuepause/noop/BUILD.bazel | 23 +++ submitqueue/extension/queuepause/noop/noop.go | 38 +++++ .../extension/queuepause/noop/noop_test.go | 34 +++++ 7 files changed, 374 insertions(+) create mode 100644 submitqueue/extension/queuepause/file/BUILD.bazel create mode 100644 submitqueue/extension/queuepause/file/file.go create mode 100644 submitqueue/extension/queuepause/file/file_test.go create mode 100644 submitqueue/extension/queuepause/noop/BUILD.bazel create mode 100644 submitqueue/extension/queuepause/noop/noop.go create mode 100644 submitqueue/extension/queuepause/noop/noop_test.go diff --git a/submitqueue/extension/queuepause/README.md b/submitqueue/extension/queuepause/README.md index 61b714d78..61e807e15 100644 --- a/submitqueue/extension/queuepause/README.md +++ b/submitqueue/extension/queuepause/README.md @@ -10,6 +10,13 @@ Returns the pause state of a queue by name. The state is two independent switche The store only reports state. The gateway and orchestrator enforce it, so every deployment runs the same rejection and freezing logic and only the source of the state differs. +## Implementations + +- `noop` never pauses a queue. It is the default when no pause state is configured. +- `file` reads a YAML file listing the queues that have switches set, and re-reads it on every `Get`, so editing the file pauses or unpauses a queue without a restart. The file must exist and parse when the store is constructed, so a mistyped path fails at startup. A later read or parse failure is returned as an error and never treated as not paused. Unknown fields are rejected so a misspelled switch cannot silently leave a queue unpaused. + +A deployer can supply its own implementation backed by a config service. + ## Entities `QueuePause` lives in `submitqueue/entity/queue_pause.go`. diff --git a/submitqueue/extension/queuepause/file/BUILD.bazel b/submitqueue/extension/queuepause/file/BUILD.bazel new file mode 100644 index 000000000..304c70370 --- /dev/null +++ b/submitqueue/extension/queuepause/file/BUILD.bazel @@ -0,0 +1,24 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["file.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/file", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + "@in_gopkg_yaml_v3//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["file_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", + ], +) diff --git a/submitqueue/extension/queuepause/file/file.go b/submitqueue/extension/queuepause/file/file.go new file mode 100644 index 000000000..65295547c --- /dev/null +++ b/submitqueue/extension/queuepause/file/file.go @@ -0,0 +1,109 @@ +// 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 file provides a YAML-file-backed implementation of queuepause.Store. +// The file is re-read on every Get, so editing it pauses or unpauses a queue +// without a restart. The file lists the queues that have pause switches set: +// +// queues: +// - name: my-queue +// reject_lands: true +// freeze_processing: false +// +// A queue with no entry is not paused. The file must exist and parse when the +// Store is constructed, so a mistyped path fails at startup. A later read or +// parse failure is returned from Get as an error, never treated as unpaused. +package file + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" + yamlv3 "gopkg.in/yaml.v3" +) + +var _ queuepause.Store = Store{} + +type fileContents struct { + Queues []queueEntry `yaml:"queues"` +} + +type queueEntry struct { + Name string `yaml:"name"` + RejectLands bool `yaml:"reject_lands"` + FreezeProcessing bool `yaml:"freeze_processing"` +} + +// Store is a queuepause.Store backed by a YAML file that is re-read on every +// Get. Safe for concurrent use. +type Store struct { + path string +} + +// NewStore returns a Store reading the YAML file at path. +// Returns an error if the file is unreadable or invalid, so that a +// misconfigured path is caught at startup. +func NewStore(path string) (Store, error) { + if _, err := load(path); err != nil { + return Store{}, err + } + return Store{path: path}, nil +} + +// Get re-reads the file and returns the pause state of the named queue, or the +// zero state if the queue has no entry. +func (s Store) Get(_ context.Context, queue string) (entity.QueuePauseState, error) { + states, err := load(s.path) + if err != nil { + return entity.QueuePauseState{}, err + } + return states[queue], nil +} + +func load(path string) (map[string]entity.QueuePauseState, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("failed to read queue pause file %q: %w", path, err) + } + + // Unknown fields are rejected: a misspelled switch would otherwise be + // ignored and leave the queue unpaused. + decoder := yamlv3.NewDecoder(bytes.NewReader(data)) + decoder.KnownFields(true) + var contents fileContents + if err := decoder.Decode(&contents); err != nil && !errors.Is(err, io.EOF) { + return nil, fmt.Errorf("failed to parse queue pause file %q: %w", path, err) + } + + states := make(map[string]entity.QueuePauseState, len(contents.Queues)) + for _, q := range contents.Queues { + if q.Name == "" { + return nil, fmt.Errorf("queue pause entry in %q has empty name", path) + } + if _, dup := states[q.Name]; dup { + return nil, fmt.Errorf("queue pause file %q has duplicate name %q", path, q.Name) + } + states[q.Name] = entity.QueuePauseState{ + RejectLands: q.RejectLands, + FreezeProcessing: q.FreezeProcessing, + } + } + return states, nil +} diff --git a/submitqueue/extension/queuepause/file/file_test.go b/submitqueue/extension/queuepause/file/file_test.go new file mode 100644 index 000000000..111983d12 --- /dev/null +++ b/submitqueue/extension/queuepause/file/file_test.go @@ -0,0 +1,139 @@ +// 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 file + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" +) + +func writeFile(t *testing.T, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "pause.yaml") + require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) + return path +} + +func TestGet(t *testing.T) { + const contents = ` +queues: + - name: rejecting + reject_lands: true + - name: freezing + freeze_processing: true + - name: both + reject_lands: true + freeze_processing: true + - name: listed-but-running +` + tests := []struct { + name string + queue string + want entity.QueuePauseState + }{ + {name: "reject only", queue: "rejecting", want: entity.QueuePauseState{RejectLands: true}}, + {name: "freeze only", queue: "freezing", want: entity.QueuePauseState{FreezeProcessing: true}}, + {name: "both switches", queue: "both", want: entity.QueuePauseState{RejectLands: true, FreezeProcessing: true}}, + {name: "listed with no switches set", queue: "listed-but-running", want: entity.QueuePauseState{}}, + {name: "unlisted queue is not paused", queue: "other", want: entity.QueuePauseState{}}, + } + store, err := NewStore(writeFile(t, contents)) + require.NoError(t, err) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := store.Get(context.Background(), tt.queue) + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestNewStoreEmptyFile(t *testing.T) { + store, err := NewStore(writeFile(t, "")) + require.NoError(t, err) + + got, err := store.Get(context.Background(), "any") + require.NoError(t, err) + assert.Equal(t, entity.QueuePauseState{}, got) +} + +func TestNewStoreRejectsInvalidFile(t *testing.T) { + tests := []struct { + name string + contents string + }{ + {name: "malformed yaml", contents: "queues: [unterminated"}, + {name: "empty queue name", contents: "queues:\n - reject_lands: true\n"}, + {name: "duplicate queue name", contents: "queues:\n - name: a\n - name: a\n"}, + {name: "misspelled switch", contents: "queues:\n - name: a\n reject_land: true\n"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := NewStore(writeFile(t, tt.contents)) + require.Error(t, err) + }) + } + + t.Run("missing file", func(t *testing.T) { + _, err := NewStore(filepath.Join(t.TempDir(), "absent.yaml")) + require.Error(t, err) + }) +} + +func TestGetReflectsFileChanges(t *testing.T) { + path := writeFile(t, "queues:\n - name: q\n") + store, err := NewStore(path) + require.NoError(t, err) + + got, err := store.Get(context.Background(), "q") + require.NoError(t, err) + assert.Equal(t, entity.QueuePauseState{}, got) + + require.NoError(t, os.WriteFile(path, []byte("queues:\n - name: q\n reject_lands: true\n"), 0o600)) + got, err = store.Get(context.Background(), "q") + require.NoError(t, err) + assert.Equal(t, entity.QueuePauseState{RejectLands: true}, got) +} + +func TestGetFailsWhenFileBecomesUnreadable(t *testing.T) { + tests := []struct { + name string + mutate func(t *testing.T, path string) + }{ + {name: "file removed", mutate: func(t *testing.T, path string) { require.NoError(t, os.Remove(path)) }}, + {name: "file corrupted", mutate: func(t *testing.T, path string) { + require.NoError(t, os.WriteFile(path, []byte("queues: [unterminated"), 0o600)) + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := writeFile(t, "queues:\n - name: q\n reject_lands: true\n") + store, err := NewStore(path) + require.NoError(t, err) + + tt.mutate(t, path) + + _, err = store.Get(context.Background(), "q") + require.Error(t, err) + }) + } +} diff --git a/submitqueue/extension/queuepause/noop/BUILD.bazel b/submitqueue/extension/queuepause/noop/BUILD.bazel new file mode 100644 index 000000000..e65643a38 --- /dev/null +++ b/submitqueue/extension/queuepause/noop/BUILD.bazel @@ -0,0 +1,23 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["noop.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["noop_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", + ], +) diff --git a/submitqueue/extension/queuepause/noop/noop.go b/submitqueue/extension/queuepause/noop/noop.go new file mode 100644 index 000000000..13e4d29eb --- /dev/null +++ b/submitqueue/extension/queuepause/noop/noop.go @@ -0,0 +1,38 @@ +// 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 noop provides a queuepause.Store under which no queue is ever paused. +package noop + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" +) + +var _ queuepause.Store = Store{} + +// Store is a queuepause.Store that reports every queue as not paused. +type Store struct{} + +// New returns a Store that never pauses a queue. +func New() Store { + return Store{} +} + +// Get always returns the zero state. +func (Store) Get(context.Context, string) (entity.QueuePauseState, error) { + return entity.QueuePauseState{}, nil +} diff --git a/submitqueue/extension/queuepause/noop/noop_test.go b/submitqueue/extension/queuepause/noop/noop_test.go new file mode 100644 index 000000000..3f6e2633b --- /dev/null +++ b/submitqueue/extension/queuepause/noop/noop_test.go @@ -0,0 +1,34 @@ +// 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 noop + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" +) + +func TestGetNeverPauses(t *testing.T) { + for _, queue := range []string{"", "any", "another"} { + t.Run(queue, func(t *testing.T) { + got, err := New().Get(context.Background(), queue) + require.NoError(t, err) + assert.Equal(t, entity.QueuePauseState{}, got) + }) + } +} From da0c4478cc212bfe963f22b0c836b0377b6cba16 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:20:14 +0000 Subject: [PATCH 2/4] refactor(queuepause): Use the renamed QueuePause entity in the stores --- submitqueue/extension/queuepause/file/file.go | 10 +++++----- .../extension/queuepause/file/file_test.go | 18 +++++++++--------- submitqueue/extension/queuepause/noop/noop.go | 4 ++-- .../extension/queuepause/noop/noop_test.go | 2 +- 4 files changed, 17 insertions(+), 17 deletions(-) diff --git a/submitqueue/extension/queuepause/file/file.go b/submitqueue/extension/queuepause/file/file.go index 65295547c..b5ce8c0f6 100644 --- a/submitqueue/extension/queuepause/file/file.go +++ b/submitqueue/extension/queuepause/file/file.go @@ -69,15 +69,15 @@ func NewStore(path string) (Store, error) { // Get re-reads the file and returns the pause state of the named queue, or the // zero state if the queue has no entry. -func (s Store) Get(_ context.Context, queue string) (entity.QueuePauseState, error) { +func (s Store) Get(_ context.Context, queue string) (entity.QueuePause, error) { states, err := load(s.path) if err != nil { - return entity.QueuePauseState{}, err + return entity.QueuePause{}, err } return states[queue], nil } -func load(path string) (map[string]entity.QueuePauseState, error) { +func load(path string) (map[string]entity.QueuePause, error) { data, err := os.ReadFile(path) if err != nil { return nil, fmt.Errorf("failed to read queue pause file %q: %w", path, err) @@ -92,7 +92,7 @@ func load(path string) (map[string]entity.QueuePauseState, error) { return nil, fmt.Errorf("failed to parse queue pause file %q: %w", path, err) } - states := make(map[string]entity.QueuePauseState, len(contents.Queues)) + states := make(map[string]entity.QueuePause, len(contents.Queues)) for _, q := range contents.Queues { if q.Name == "" { return nil, fmt.Errorf("queue pause entry in %q has empty name", path) @@ -100,7 +100,7 @@ func load(path string) (map[string]entity.QueuePauseState, error) { if _, dup := states[q.Name]; dup { return nil, fmt.Errorf("queue pause file %q has duplicate name %q", path, q.Name) } - states[q.Name] = entity.QueuePauseState{ + states[q.Name] = entity.QueuePause{ RejectLands: q.RejectLands, FreezeProcessing: q.FreezeProcessing, } diff --git a/submitqueue/extension/queuepause/file/file_test.go b/submitqueue/extension/queuepause/file/file_test.go index 111983d12..fea11ac8c 100644 --- a/submitqueue/extension/queuepause/file/file_test.go +++ b/submitqueue/extension/queuepause/file/file_test.go @@ -47,13 +47,13 @@ queues: tests := []struct { name string queue string - want entity.QueuePauseState + want entity.QueuePause }{ - {name: "reject only", queue: "rejecting", want: entity.QueuePauseState{RejectLands: true}}, - {name: "freeze only", queue: "freezing", want: entity.QueuePauseState{FreezeProcessing: true}}, - {name: "both switches", queue: "both", want: entity.QueuePauseState{RejectLands: true, FreezeProcessing: true}}, - {name: "listed with no switches set", queue: "listed-but-running", want: entity.QueuePauseState{}}, - {name: "unlisted queue is not paused", queue: "other", want: entity.QueuePauseState{}}, + {name: "reject only", queue: "rejecting", want: entity.QueuePause{RejectLands: true}}, + {name: "freeze only", queue: "freezing", want: entity.QueuePause{FreezeProcessing: true}}, + {name: "both switches", queue: "both", want: entity.QueuePause{RejectLands: true, FreezeProcessing: true}}, + {name: "listed with no switches set", queue: "listed-but-running", want: entity.QueuePause{}}, + {name: "unlisted queue is not paused", queue: "other", want: entity.QueuePause{}}, } store, err := NewStore(writeFile(t, contents)) require.NoError(t, err) @@ -73,7 +73,7 @@ func TestNewStoreEmptyFile(t *testing.T) { got, err := store.Get(context.Background(), "any") require.NoError(t, err) - assert.Equal(t, entity.QueuePauseState{}, got) + assert.Equal(t, entity.QueuePause{}, got) } func TestNewStoreRejectsInvalidFile(t *testing.T) { @@ -106,12 +106,12 @@ func TestGetReflectsFileChanges(t *testing.T) { got, err := store.Get(context.Background(), "q") require.NoError(t, err) - assert.Equal(t, entity.QueuePauseState{}, got) + assert.Equal(t, entity.QueuePause{}, got) require.NoError(t, os.WriteFile(path, []byte("queues:\n - name: q\n reject_lands: true\n"), 0o600)) got, err = store.Get(context.Background(), "q") require.NoError(t, err) - assert.Equal(t, entity.QueuePauseState{RejectLands: true}, got) + assert.Equal(t, entity.QueuePause{RejectLands: true}, got) } func TestGetFailsWhenFileBecomesUnreadable(t *testing.T) { diff --git a/submitqueue/extension/queuepause/noop/noop.go b/submitqueue/extension/queuepause/noop/noop.go index 13e4d29eb..99054da71 100644 --- a/submitqueue/extension/queuepause/noop/noop.go +++ b/submitqueue/extension/queuepause/noop/noop.go @@ -33,6 +33,6 @@ func New() Store { } // Get always returns the zero state. -func (Store) Get(context.Context, string) (entity.QueuePauseState, error) { - return entity.QueuePauseState{}, nil +func (Store) Get(context.Context, string) (entity.QueuePause, error) { + return entity.QueuePause{}, nil } diff --git a/submitqueue/extension/queuepause/noop/noop_test.go b/submitqueue/extension/queuepause/noop/noop_test.go index 3f6e2633b..c8924a650 100644 --- a/submitqueue/extension/queuepause/noop/noop_test.go +++ b/submitqueue/extension/queuepause/noop/noop_test.go @@ -28,7 +28,7 @@ func TestGetNeverPauses(t *testing.T) { t.Run(queue, func(t *testing.T) { got, err := New().Get(context.Background(), queue) require.NoError(t, err) - assert.Equal(t, entity.QueuePauseState{}, got) + assert.Equal(t, entity.QueuePause{}, got) }) } } From de7e380efd48d27709b4b2dce43669da2ae14921 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:23:42 +0000 Subject: [PATCH 3/4] feat(queuepause): Add an in-memory fake store for tests --- submitqueue/extension/queuepause/README.md | 2 + .../extension/queuepause/fake/BUILD.bazel | 23 +++++++ submitqueue/extension/queuepause/fake/fake.go | 69 +++++++++++++++++++ .../extension/queuepause/fake/fake_test.go | 64 +++++++++++++++++ 4 files changed, 158 insertions(+) create mode 100644 submitqueue/extension/queuepause/fake/BUILD.bazel create mode 100644 submitqueue/extension/queuepause/fake/fake.go create mode 100644 submitqueue/extension/queuepause/fake/fake_test.go diff --git a/submitqueue/extension/queuepause/README.md b/submitqueue/extension/queuepause/README.md index 61e807e15..7e3b7f8a9 100644 --- a/submitqueue/extension/queuepause/README.md +++ b/submitqueue/extension/queuepause/README.md @@ -15,6 +15,8 @@ The store only reports state. The gateway and orchestrator enforce it, so every - `noop` never pauses a queue. It is the default when no pause state is configured. - `file` reads a YAML file listing the queues that have switches set, and re-reads it on every `Get`, so editing the file pauses or unpauses a queue without a restart. The file must exist and parse when the store is constructed, so a mistyped path fails at startup. A later read or parse failure is returned as an error and never treated as not paused. Unknown fields are rejected so a misspelled switch cannot silently leave a queue unpaused. +- `fake` is an in-memory store for tests. A test sets a queue's state, changes it while consumers run, and can make every read fail. It is never used in production. + A deployer can supply its own implementation backed by a config service. ## Entities diff --git a/submitqueue/extension/queuepause/fake/BUILD.bazel b/submitqueue/extension/queuepause/fake/BUILD.bazel new file mode 100644 index 000000000..5e2d56b30 --- /dev/null +++ b/submitqueue/extension/queuepause/fake/BUILD.bazel @@ -0,0 +1,23 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["fake.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/fake", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["fake_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", + ], +) diff --git a/submitqueue/extension/queuepause/fake/fake.go b/submitqueue/extension/queuepause/fake/fake.go new file mode 100644 index 000000000..2cb0d1e30 --- /dev/null +++ b/submitqueue/extension/queuepause/fake/fake.go @@ -0,0 +1,69 @@ +// 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 fake provides an in-memory queuepause.Store for tests and examples. +// Seed or change a queue's state with Set; a queue never set is not paused. +// FailWith injects a read error on every Get to exercise the error path without +// a real backend. Safe for concurrent use, so a test can change state while a +// consumer is reading. It is intended for tests and examples only, never +// production. +package fake + +import ( + "context" + "sync" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" +) + +var _ queuepause.Store = (*Store)(nil) + +// Store is a programmable in-memory queuepause.Store. +type Store struct { + mu sync.RWMutex + states map[string]entity.QueuePause + err error +} + +// New returns a Store under which no queue is paused. +func New() *Store { + return &Store{states: map[string]entity.QueuePause{}} +} + +// Set replaces the pause state of a queue. Setting the zero state unpauses it. +func (s *Store) Set(queue string, state entity.QueuePause) *Store { + s.mu.Lock() + defer s.mu.Unlock() + s.states[queue] = state + return s +} + +// FailWith makes every Get return err. Passing nil restores normal reads. +func (s *Store) FailWith(err error) *Store { + s.mu.Lock() + defer s.mu.Unlock() + s.err = err + return s +} + +// Get returns the state last set for the queue, or the zero state if none was. +func (s *Store) Get(_ context.Context, queue string) (entity.QueuePause, error) { + s.mu.RLock() + defer s.mu.RUnlock() + if s.err != nil { + return entity.QueuePause{}, s.err + } + return s.states[queue], nil +} diff --git a/submitqueue/extension/queuepause/fake/fake_test.go b/submitqueue/extension/queuepause/fake/fake_test.go new file mode 100644 index 000000000..8198dc389 --- /dev/null +++ b/submitqueue/extension/queuepause/fake/fake_test.go @@ -0,0 +1,64 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fake + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +func TestGet(t *testing.T) { + frozen := entity.QueuePause{FreezeProcessing: true, Reason: "incident"} + tests := []struct { + name string + setup func(*Store) + queue string + want entity.QueuePause + }{ + {name: "queue never set is not paused", setup: func(*Store) {}, queue: "q", want: entity.QueuePause{}}, + {name: "queue set is returned", setup: func(s *Store) { s.Set("q", frozen) }, queue: "q", want: frozen}, + {name: "other queue is unaffected", setup: func(s *Store) { s.Set("q", frozen) }, queue: "other", want: entity.QueuePause{}}, + {name: "setting the zero state unpauses", setup: func(s *Store) { s.Set("q", frozen).Set("q", entity.QueuePause{}) }, queue: "q", want: entity.QueuePause{}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := New() + tt.setup(s) + + got, err := s.Get(context.Background(), tt.queue) + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestFailWith(t *testing.T) { + boom := errors.New("backend down") + s := New().Set("q", entity.QueuePause{RejectLands: true}).FailWith(boom) + + got, err := s.Get(context.Background(), "q") + require.ErrorIs(t, err, boom) + assert.Equal(t, entity.QueuePause{}, got) + + got, err = s.FailWith(nil).Get(context.Background(), "q") + require.NoError(t, err) + assert.Equal(t, entity.QueuePause{RejectLands: true}, got) +} From 76bc41b8fd6ff764e02313f4eca3d76d3ef6c014 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:24:48 +0000 Subject: [PATCH 4/4] feat(queuepause): Read an optional reason for each queue in the file store --- submitqueue/extension/queuepause/file/file.go | 3 +++ submitqueue/extension/queuepause/file/file_test.go | 4 ++++ 2 files changed, 7 insertions(+) diff --git a/submitqueue/extension/queuepause/file/file.go b/submitqueue/extension/queuepause/file/file.go index b5ce8c0f6..c49bdfe9e 100644 --- a/submitqueue/extension/queuepause/file/file.go +++ b/submitqueue/extension/queuepause/file/file.go @@ -20,6 +20,7 @@ // - name: my-queue // reject_lands: true // freeze_processing: false +// reason: repo freeze until 17:00 // // A queue with no entry is not paused. The file must exist and parse when the // Store is constructed, so a mistyped path fails at startup. A later read or @@ -49,6 +50,7 @@ type queueEntry struct { Name string `yaml:"name"` RejectLands bool `yaml:"reject_lands"` FreezeProcessing bool `yaml:"freeze_processing"` + Reason string `yaml:"reason"` } // Store is a queuepause.Store backed by a YAML file that is re-read on every @@ -103,6 +105,7 @@ func load(path string) (map[string]entity.QueuePause, error) { states[q.Name] = entity.QueuePause{ RejectLands: q.RejectLands, FreezeProcessing: q.FreezeProcessing, + Reason: q.Reason, } } return states, nil diff --git a/submitqueue/extension/queuepause/file/file_test.go b/submitqueue/extension/queuepause/file/file_test.go index fea11ac8c..0981b7e1a 100644 --- a/submitqueue/extension/queuepause/file/file_test.go +++ b/submitqueue/extension/queuepause/file/file_test.go @@ -42,6 +42,9 @@ queues: - name: both reject_lands: true freeze_processing: true + - name: with-reason + reject_lands: true + reason: repo freeze until 17:00 - name: listed-but-running ` tests := []struct { @@ -52,6 +55,7 @@ queues: {name: "reject only", queue: "rejecting", want: entity.QueuePause{RejectLands: true}}, {name: "freeze only", queue: "freezing", want: entity.QueuePause{FreezeProcessing: true}}, {name: "both switches", queue: "both", want: entity.QueuePause{RejectLands: true, FreezeProcessing: true}}, + {name: "reason is carried", queue: "with-reason", want: entity.QueuePause{RejectLands: true, Reason: "repo freeze until 17:00"}}, {name: "listed with no switches set", queue: "listed-but-running", want: entity.QueuePause{}}, {name: "unlisted queue is not paused", queue: "other", want: entity.QueuePause{}}, }