feat(stovepipe): add queueconfig Store with default implementation

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
 }
@@ -51,13 +53,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,
 	}
@@ -101,8 +105,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 {
@@ -134,6 +138,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"
@@ -57,7 +58,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
 }
 
@@ -131,6 +132,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:   "request/monorepo/main/3",

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

stovepipe/extension/queueconfig/default/default.go

@@ -0,0 +1,55 @@
+// 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 default provides a queueconfig.Store that returns global wiring
+// defaults for any non-empty queue name. Replace with a YAML- or service-backed
+// store when per-queue overrides are needed.
+package defaultconfig
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
+)
+
+const (
+	_defaultMaxConcurrent   = 1
+	_defaultGateWaitDelayMs = 5000
+)
+
+// Store is a queueconfig.Store that returns the same defaults for every queue.
+type Store struct{}
+
+// NewStore returns a Store backed by global wiring defaults.
+func NewStore() Store {
+	return Store{}
+}
+
+// Get returns the default configuration for any non-empty queue name.
+func (Store) Get(_ context.Context, name string) (entity.QueueConfig, error) {
+	if name == "" {
+		return entity.QueueConfig{}, queueconfig.ErrNotFound
+	}
+	return entity.QueueConfig{
+		Name:            name,
+		MaxConcurrent:   _defaultMaxConcurrent,
+		GateWaitDelayMs: _defaultGateWaitDelayMs,
+	}, nil
+}
+
+// List returns an empty slice; this store does not enumerate known queues.
+func (Store) List(context.Context) ([]entity.QueueConfig, error) {
+	return nil, nil
+}

stovepipe/extension/queueconfig/default/default_test.go

@@ -0,0 +1,49 @@
+// 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 defaultconfig
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
+)
+
+func TestStore_Get(t *testing.T) {
+	store := NewStore()
+
+	t.Run("returns defaults for named queue", func(t *testing.T) {
+		cfg, err := store.Get(context.Background(), "monorepo/main")
+		require.NoError(t, err)
+		assert.Equal(t, "monorepo/main", cfg.Name)
+		assert.Equal(t, int32(1), cfg.MaxConcurrent)
+		assert.Equal(t, int64(5000), cfg.GateWaitDelayMs)
+	})
+
+	t.Run("empty name is not found", func(t *testing.T) {
+		_, err := store.Get(context.Background(), "")
+		require.ErrorIs(t, err, queueconfig.ErrNotFound)
+	})
+}
+
+func TestStore_List(t *testing.T) {
+	store := NewStore()
+
+	cfg, err := store.List(context.Background())
+	require.NoError(t, err)
+	assert.Empty(t, cfg)
+}

stovepipe/extension/queueconfig/mock/BUILD.bazel

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