feat(submitqueue): Add noop, file, fake queuepause store

submitqueue/extension/queuepause/README.md

@@ -10,6 +10,15 @@ 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.
+
+- `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
 
 `QueuePause` lives in `submitqueue/entity/queue_pause.go`.

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

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

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

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

submitqueue/extension/queuepause/file/file.go

@@ -0,0 +1,112 @@
+// 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
+//	    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
+// 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"`
+	Reason           string `yaml:"reason"`
+}
+
+// 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.QueuePause, error) {
+	states, err := load(s.path)
+	if err != nil {
+		return entity.QueuePause{}, err
+	}
+	return states[queue], nil
+}
+
+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)
+	}
+
+	// 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.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)
+		}
+		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.QueuePause{
+			RejectLands:      q.RejectLands,
+			FreezeProcessing: q.FreezeProcessing,
+			Reason:           q.Reason,
+		}
+	}
+	return states, nil
+}