feat(stovepipe): per-queue config yaml

doc/rfc/stovepipe/steps/process.md

@@ -62,7 +62,7 @@ Validation is expensive and shares a baseline, so heads arriving while an earlie
 | Queue row | `last_green_uri` | Bookmark `record` advances on whole-repo green; empty until first green. |
 | Queue row | `in_flight_count` | Requests past `process` and not yet terminal. `process` increments on admit; `buildsignal` (or DLQ reconciliation) decrements on terminal. |
 | Queue row | `build_admission_not_before_ms` | Durable earliest time for the next logical admission; zero until a time policy advances it. |
-| Queue config | `max_concurrent` | Cap on concurrent in-flight validations. **Default 1** (global wiring default for MVP; per-queue override when a Stovepipe `queueconfig` extension lands). |
+| Queue config | `max_concurrent` | Cap on concurrent in-flight validations. **Default 1**; the YAML store supports per-queue overrides. |
 | Queue config | `minimum_build_admission_interval_ms` | Minimum start-to-start spacing between logical admissions. Positive values enable general throttling; non-positive values disable it. **Default 0**. |
 
 A slot is held from admit until the build goes terminal (`process → build → buildsignal`), not just while `process` runs. It is released when the Request reaches **any** terminal state and `in_flight_count` is decremented — `buildsignal` recording the build's outcome, success *or* failure, or the DLQ reconciler forcing a terminal `failed` (see [integrity](#in_flight_count-integrity)). A build *failure* frees the slot just like a success; only a Request that never terminates keeps its slot.

service/stovepipe/README.md

@@ -24,6 +24,8 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `queu
 - **`inMemoryCounter`** — a process-local `counter.Counter` for sequence numbers; not durable. A real deployment uses a persistent implementation (e.g. `platform/extension/counter/mysql`).
 - **`fakeSourceControlFactory`** — seeds each queue with a deterministic single-commit history so ingest resolves a stable head URI (and re-ingesting the same queue exercises the dedup path). A real deployment supplies a VCS-backed `sourcecontrol.Factory`, which is also where a queue's promotion ref is resolved. The fake has no ref to move, so a promotion locally shows up only in the record consumer's logs.
 
+`QUEUE_CONFIG_PATH` selects the YAML-backed queue policy store. When it is unset, the server uses the built-in defaults (`max_concurrent: 1`, `gate_wait_delay_ms: 5000`, and time-based admission throttling disabled). When it is set, the configured queue names must exactly match `MQ_TENANTS`. Each configured queue can set `minimum_build_admission_interval_ms` for general start-to-start throttling. For example, `minimum_build_admission_interval_ms: 3600000` permits at most one logical admission per hour.
+
 ## Layout
 
 ```
@@ -46,6 +48,7 @@ The Stovepipe controllers live under [`stovepipe/controller/`](../../stovepipe/c
 | `STORAGE_MYSQL_DSN` | yes      | Storage database DSN | — |
 | `QUEUE_MYSQL_DSN`   | yes      | Queue database DSN                       | —                    |
 | `QUEUE_LOG_LEVEL`   | no       | Message-queue logger level               | `info`               |
+| `QUEUE_CONFIG_PATH` | no       | YAML file containing per-queue admission policies | built-in defaults |
 | `PORT`              | no       | gRPC listen address                      | `:8083`              |
 | `HOSTNAME`          | no       | Subscriber name for the queue consumers  | `stovepipe-<unix_ts>` |
 

service/stovepipe/docker-compose.yml

@@ -71,6 +71,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:-}
+      - QUEUE_CONFIG_PATH=/root/queues.yaml
       - MQ_TENANTS=${MQ_TENANTS:-monorepo/main,monorepo/release,monorepo/slow?buildrunner-fake=build-slow}
       - HOSTNAME=stovepipe-dev
     depends_on:

service/stovepipe/server/BUILD.bazel

@@ -33,7 +33,9 @@ go_library(
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/extension/buildrunner:go_default_library",
         "//stovepipe/extension/buildrunner/fake:go_default_library",
+        "//stovepipe/extension/queueconfig:go_default_library",
         "//stovepipe/extension/queueconfig/default:go_default_library",
+        "//stovepipe/extension/queueconfig/yaml:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/sourcecontrol/fake:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
@@ -69,6 +71,7 @@ filegroup(
     testonly = True,
     srcs = [
         "Dockerfile",
+        "queues.yaml",
         ":stovepipe_linux",
     ],
     visibility = ["//test:__subpackages__"],
@@ -86,6 +89,7 @@ go_test(
         "//stovepipe/controller/dlq:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/queueconfig/default: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",

service/stovepipe/server/Dockerfile

@@ -6,6 +6,7 @@ WORKDIR /root/
 # Copy pre-built Linux binary.
 # Built via: make build-stovepipe-linux
 COPY .docker-bin/stovepipe ./stovepipe
+COPY service/stovepipe/server/queues.yaml ./queues.yaml
 
 EXPOSE 8080
 

service/stovepipe/server/main.go

@@ -55,7 +55,9 @@ import (
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
 	buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
+	"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
 	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
+	queueconfigyaml "github.com/uber/submitqueue/stovepipe/extension/queueconfig/yaml"
 	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	sourcecontrolfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
@@ -330,10 +332,18 @@ func run() error {
 	// silently duplicated across controllers.
 	sourceControl := fakeSourceControlFactory{}
 	brf := fakeBuildRunnerFactory{}
+	queueConfigPath := os.Getenv("QUEUE_CONFIG_PATH")
+	queueConfigs, err := loadQueueConfigs(queueConfigPath)
+	if err != nil {
+		return err
+	}
+	if err := validateQueueConfigTenants(ctx, queueConfigPath, queueConfigs, tenants); err != nil {
+		return err
+	}
 
 	storageFty := storageFactory{backend: store}
 	materializer := requestlog.NewMaterializer(scope)
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{})
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, queueConfigs, registry, sourceControl, brf, hookResolver{})
 	if err != nil {
 		return err
 	}
@@ -449,6 +459,7 @@ func registerPrimaryControllers(
 	scope tally.Scope,
 	store storage.Factory,
 	materializer requestlog.Materializer,
+	queueConfigs queueconfig.Store,
 	registry consumer.TopicRegistry,
 	sourceControl sourcecontrol.Factory,
 	brf buildrunner.Factory,
@@ -499,6 +510,35 @@ func registerPrimaryControllers(
 	return count, nil
 }
 
+func loadQueueConfigs(path string) (queueconfig.Store, error) {
+	if path == "" {
+		return queueconfigdefault.NewStore(), nil
+	}
+	store, err := queueconfigyaml.NewStore(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load queue configs: %w", err)
+	}
+	return store, nil
+}
+
+func validateQueueConfigTenants(ctx context.Context, path string, store queueconfig.Store, tenants []string) error {
+	if path == "" {
+		return nil
+	}
+	configs, err := store.List(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to list queue configs: %w", err)
+	}
+	configuredQueueNames := make([]string, 0, len(configs))
+	for _, cfg := range configs {
+		configuredQueueNames = append(configuredQueueNames, cfg.Name)
+	}
+	if err := servicemq.ValidateTenantSetsEqual("MQ_TENANTS", tenants, "QUEUE_CONFIG_PATH", configuredQueueNames); err != nil {
+		return fmt.Errorf("failed to validate queue config tenants: %w", err)
+	}
+	return nil
+}
+
 // registerDLQControllers creates one DLQ reconciler per primary stage and
 // registers them with c, returning how many were registered.
 func registerDLQControllers(

service/stovepipe/server/main_test.go

@@ -17,6 +17,8 @@ package main
 import (
 	"context"
 	"errors"
+	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 
@@ -30,6 +32,7 @@ import (
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -173,7 +176,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con
 	primary := &recordingConsumer{}
 	deadLetter := &recordingConsumer{}
 
-	_, err = registerPrimaryControllers(primary, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry,
+	_, err = registerPrimaryControllers(primary, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), queueconfigdefault.NewStore(), registry,
 		fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{})
 	require.NoError(t, err)
 
@@ -227,3 +230,39 @@ func TestHookStage(t *testing.T) {
 		}
 	})
 }
+
+func TestLoadQueueConfigs(t *testing.T) {
+	t.Run("empty path uses defaults", func(t *testing.T) {
+		store, err := loadQueueConfigs("")
+		require.NoError(t, err)
+
+		cfg, err := store.Get(context.Background(), "monorepo/main")
+		require.NoError(t, err)
+		assert.EqualValues(t, 1, cfg.MaxConcurrent)
+		assert.Zero(t, cfg.MinimumBuildAdmissionIntervalMs)
+	})
+
+	t.Run("path loads per-queue policies", func(t *testing.T) {
+		path := filepath.Join(t.TempDir(), "queues.yaml")
+		require.NoError(t, os.WriteFile(path, []byte(`queues:
+  - name: monorepo/main
+    max_concurrent: 2
+    gate_wait_delay_ms: 5000
+    minimum_build_admission_interval_ms: 3600000
+`), 0o600))
+
+		store, err := loadQueueConfigs(path)
+		require.NoError(t, err)
+		cfg, err := store.Get(context.Background(), "monorepo/main")
+		require.NoError(t, err)
+		assert.EqualValues(t, 2, cfg.MaxConcurrent)
+		assert.EqualValues(t, 3_600_000, cfg.MinimumBuildAdmissionIntervalMs)
+		require.NoError(t, validateQueueConfigTenants(context.Background(), path, store, []string{"monorepo/main"}))
+		require.Error(t, validateQueueConfigTenants(context.Background(), path, store, []string{"other"}))
+	})
+
+	t.Run("invalid path fails", func(t *testing.T) {
+		_, err := loadQueueConfigs(filepath.Join(t.TempDir(), "missing.yaml"))
+		require.Error(t, err)
+	})
+}

service/stovepipe/server/queues.yaml

@@ -0,0 +1,15 @@
+# Queue policies for the local Stovepipe stack. Durations are milliseconds.
+queues:
+  - name: monorepo/main
+    max_concurrent: 1
+    gate_wait_delay_ms: 5000
+    # Set to 3600000 to admit at most one build per hour.
+    minimum_build_admission_interval_ms: 0
+  - name: monorepo/release
+    max_concurrent: 1
+    gate_wait_delay_ms: 5000
+    minimum_build_admission_interval_ms: 0
+  - name: "monorepo/slow?buildrunner-fake=build-slow"
+    max_concurrent: 1
+    gate_wait_delay_ms: 5000
+    minimum_build_admission_interval_ms: 0

stovepipe/extension/queueconfig/README.md

@@ -10,8 +10,9 @@ Pipeline stages read mutable runtime state from storage and read knobs such as `
 
 ## Entities
 
-Queue configuration entity lives in `stovepipe/entity/queue_config.go` and carries deployment knobs (`max_concurrent`, `gate_wait_delay_ms`, `minimum_build_admission_interval_ms`) separate from the mutable `Queue` row. The minimum interval is start-to-start spacing between logical admissions; zero disables time-based throttling.
+Queue configuration entity lives in `stovepipe/entity/queue_config.go` and carries deployment knobs (`max_concurrent`, `gate_wait_delay_ms`, `minimum_build_admission_interval_ms`) separate from the mutable `Queue` row. The minimum interval is start-to-start spacing between logical admissions. Positive values enable the policy; non-positive values disable it.
 
 ## Implementations
 
-`default` returns the global wiring defaults for any non-empty queue name until a file- or service-backed store lands.
+- `default` returns the global wiring defaults for any non-empty queue name. Time-based admission throttling is disabled.
+- `yaml` loads a validated immutable snapshot from a file. Every queue entry specifies its concurrency, gate delay, and general minimum admission interval.

stovepipe/extension/queueconfig/yaml/BUILD.bazel

@@ -0,0 +1,25 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["yaml.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/extension/queueconfig/yaml",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/queueconfig:go_default_library",
+        "@in_gopkg_yaml_v3//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["yaml_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/queueconfig:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)

stovepipe/extension/queueconfig/yaml/yaml.go

@@ -0,0 +1,96 @@
+// 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 yaml provides a YAML-file-backed queueconfig.Store.
+package yaml
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
+	yamlv3 "gopkg.in/yaml.v3"
+)
+
+type fileContents struct {
+	Queues []entity.QueueConfig `yaml:"queues"`
+}
+
+// Store is an immutable in-memory snapshot of a YAML queue configuration file.
+type Store struct {
+	byName map[string]entity.QueueConfig
+	all    []entity.QueueConfig
+}
+
+// NewStore reads and validates queue configurations from path.
+func NewStore(path string) (Store, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return Store{}, fmt.Errorf("failed to read queue config file %q: %w", path, err)
+	}
+
+	var contents fileContents
+	decoder := yamlv3.NewDecoder(bytes.NewReader(data))
+	decoder.KnownFields(true)
+	if err := decoder.Decode(&contents); err != nil {
+		return Store{}, fmt.Errorf("failed to parse queue config file %q: %w", path, err)
+	}
+
+	byName := make(map[string]entity.QueueConfig, len(contents.Queues))
+	for _, cfg := range contents.Queues {
+		if err := validate(cfg); err != nil {
+			return Store{}, fmt.Errorf("invalid queue config in %q: %w", path, err)
+		}
+		if _, exists := byName[cfg.Name]; exists {
+			return Store{}, fmt.Errorf("queue config in %q has duplicate name %q", path, cfg.Name)
+		}
+		byName[cfg.Name] = cfg
+	}
+
+	all := make([]entity.QueueConfig, len(contents.Queues))
+	copy(all, contents.Queues)
+	return Store{byName: byName, all: all}, nil
+}
+
+func validate(cfg entity.QueueConfig) error {
+	if cfg.Name == "" {
+		return fmt.Errorf("name must not be empty")
+	}
+	if cfg.MaxConcurrent <= 0 {
+		return fmt.Errorf("max_concurrent for queue %q must be positive", cfg.Name)
+	}
+	if cfg.GateWaitDelayMs <= 0 {
+		return fmt.Errorf("gate_wait_delay_ms for queue %q must be positive", cfg.Name)
+	}
+	return nil
+}
+
+// Get returns the configuration for name.
+func (s Store) Get(_ context.Context, name string) (entity.QueueConfig, error) {
+	cfg, ok := s.byName[name]
+	if !ok {
+		return entity.QueueConfig{}, queueconfig.ErrNotFound
+	}
+	return cfg, nil
+}
+
+// List returns a copy of all configurations in file order.
+func (s Store) List(context.Context) ([]entity.QueueConfig, error) {
+	configs := make([]entity.QueueConfig, len(s.all))
+	copy(configs, s.all)
+	return configs, nil
+}