feat(stovepipe): add queueconfig Store with default implementation

Introduce queueconfig.Store and wire process to resolve per-queue
max_concurrent at gate-check time. The default store returns global wiring
constants; admit lands in a follow-up PR.

Author: mnoah1

service/stovepipe/server/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "//stovepipe/controller:go_default_library",
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/extension/queueconfig/default:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/sourcecontrol/fake:go_default_library",
         "//stovepipe/extension/storage/mysql:go_default_library",

service/stovepipe/server/main.go

@@ -38,6 +38,7 @@ import (
 	"github.com/uber/submitqueue/stovepipe/controller"
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
 	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	sourcecontrolfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake"
 	storageMySQL "github.com/uber/submitqueue/stovepipe/extension/storage/mysql"
@@ -209,7 +210,7 @@ func run() error {
 		),
 	)
 
-	processController := process.NewController(logger.Sugar(), scope, store, stovepipemq.TopicKeyProcess, "stovepipe-process")
+	processController := process.NewController(logger.Sugar(), scope, store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process")
 	if err := primaryConsumer.Register(processController); err != nil {
 		return fmt.Errorf("failed to register process controller: %w", err)
 	}

stovepipe/controller/process/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "//platform/metrics:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/queueconfig:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
@@ -28,6 +29,7 @@ go_test(
         "//platform/extension/messagequeue/mock:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/queueconfig/default:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "//stovepipe/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",

stovepipe/controller/process/process.go

@@ -29,6 +29,7 @@ import (
 	"github.com/uber/submitqueue/platform/metrics"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
 )
@@ -39,6 +40,7 @@ type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	store         storage.Storage
+	queueConfigs  queueconfig.Store
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -54,13 +56,15 @@ func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	store storage.Storage,
+	queueConfigs queueconfig.Store,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
 	return &Controller{
 		logger:        logger.Named("process_controller"),
 		metricsScope:  scope.SubScope("process_controller"),
 		store:         store,
+		queueConfigs:  queueConfigs,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
@@ -105,8 +109,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	}
 }
 
-// processAccepted coalesces older heads against queue.latest_request_id. The latest
-// head is left in accepted until admit and the concurrency gate land in later PRs.
+// processAccepted coalesces older heads against queue.latest_request_id, then resolves
+// per-queue config for the concurrency gate. Admit lands in a follow-up PR.
 func (c *Controller) processAccepted(ctx context.Context, request entity.Request) error {
 	queueRow, err := c.loadQueue(ctx, request.Queue)
 	if err != nil {
@@ -143,6 +147,20 @@ func (c *Controller) processAccepted(ctx context.Context, request entity.Request
 		return nil
 	}
 
+	cfg, err := c.queueConfigs.Get(ctx, request.Queue)
+	if err != nil {
+		return fmt.Errorf("ProcessController failed to load queue config for %s: %w", request.Queue, err)
+	}
+
+	if queueRow.InFlightCount >= cfg.MaxConcurrent {
+		c.logger.Infow("latest head awaiting build slot",
+			"request_id", request.ID,
+			"queue", request.Queue,
+			"in_flight_count", queueRow.InFlightCount,
+		)
+		return nil
+	}
+
 	c.logger.Infow("latest head awaiting admit",
 		"request_id", request.ID,
 		"queue", request.Queue,

stovepipe/controller/process/process_test.go

@@ -28,6 +28,7 @@ import (
 	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
 	"go.uber.org/mock/gomock"
@@ -58,7 +59,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processM
 	store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
 	store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
 
-	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, stovepipemq.TopicKeyProcess, "stovepipe-process")
+	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process")
 	return c, m
 }
 
@@ -141,6 +142,18 @@ func TestProcess(t *testing.T) {
 				}, nil)
 			},
 		},
+		{
+			name: "latest accepted head awaits slot when gate closed",
+			setup: func(m processMocks) {
+				m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil)
+				m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{
+					Name:            testQueue,
+					LatestRequestID: testID,
+					InFlightCount:   1,
+					Version:         1,
+				}, nil)
+			},
+		},
 		{
 			name: "older accepted head is superseded",
 			id:   testOlderID,

stovepipe/entity/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "queue.go",
+        "queue_config.go",
         "request.go",
         "request_id.go",
     ],

stovepipe/entity/queue_config.go

@@ -0,0 +1,29 @@
+// Copyright (c) 2025 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 entity
+
+// QueueConfig holds deployment configuration for a Stovepipe validation queue.
+// Mutable runtime state (latest head, in-flight count) lives on the Queue row;
+// knobs such as max_concurrent are resolved here at gate-check time. Immutable
+// after load.
+type QueueConfig struct {
+	// Name uniquely identifies this queue within the system.
+	// Referenced by Request.Queue.
+	Name string `json:"name" yaml:"name"`
+	// MaxConcurrent is the cap on concurrent in-flight validations for the queue.
+	MaxConcurrent int32 `json:"max_concurrent" yaml:"max_concurrent"`
+	// GateWaitDelayMs is the PublishAfter delay when the latest head waits for a slot.
+	GateWaitDelayMs int64 `json:"gate_wait_delay_ms" yaml:"gate_wait_delay_ms"`
+}

stovepipe/extension/queueconfig/BUILD.bazel

@@ -0,0 +1,9 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["queueconfig.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/extension/queueconfig",
+    visibility = ["//visibility:public"],
+    deps = ["//stovepipe/entity:go_default_library"],
+)

stovepipe/extension/queueconfig/README.md

@@ -0,0 +1,17 @@
+# Queue Config Extension
+
+Vendor-agnostic interface for providing Stovepipe queue configurations.
+
+Pipeline stages read mutable runtime state from storage and read knobs such as `max_concurrent` from a config `Store` at call time, matching the SubmitQueue split documented in [submitqueue/extension/queueconfig/README.md](../../../submitqueue/extension/queueconfig/README.md).
+
+## Interfaces
+
+`Store` provides queue configurations by name via `Get` and `List`. See `queueconfig.go`.
+
+## Entities
+
+Queue configuration entity lives in `stovepipe/entity/queue_config.go` and carries deployment knobs (`max_concurrent`, `gate_wait_delay_ms`) separate from the mutable `Queue` row.
+
+## Implementations
+
+`default` returns the global wiring defaults for any non-empty queue name until a file- or service-backed store lands.

stovepipe/extension/queueconfig/default/BUILD.bazel

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