From 98cef3f2ef5e2925da0d611e2427ec50da5069cc Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 10 Jul 2026 16:18:40 +0000 Subject: [PATCH] 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. --- service/stovepipe/server/BUILD.bazel | 1 + service/stovepipe/server/main.go | 3 +- stovepipe/controller/process/BUILD.bazel | 2 + stovepipe/controller/process/process.go | 22 +++++- stovepipe/controller/process/process_test.go | 15 +++- stovepipe/entity/BUILD.bazel | 1 + stovepipe/entity/queue_config.go | 29 ++++++++ stovepipe/extension/queueconfig/BUILD.bazel | 9 +++ stovepipe/extension/queueconfig/README.md | 17 +++++ .../extension/queueconfig/default/BUILD.bazel | 23 ++++++ .../extension/queueconfig/default/default.go | 55 ++++++++++++++ .../queueconfig/default/default_test.go | 49 +++++++++++++ .../extension/queueconfig/mock/BUILD.bazel | 12 ++++ .../queueconfig/mock/queueconfig_mock.go | 72 +++++++++++++++++++ .../extension/queueconfig/queueconfig.go | 38 ++++++++++ 15 files changed, 344 insertions(+), 4 deletions(-) create mode 100644 stovepipe/entity/queue_config.go create mode 100644 stovepipe/extension/queueconfig/BUILD.bazel create mode 100644 stovepipe/extension/queueconfig/README.md create mode 100644 stovepipe/extension/queueconfig/default/BUILD.bazel create mode 100644 stovepipe/extension/queueconfig/default/default.go create mode 100644 stovepipe/extension/queueconfig/default/default_test.go create mode 100644 stovepipe/extension/queueconfig/mock/BUILD.bazel create mode 100644 stovepipe/extension/queueconfig/mock/queueconfig_mock.go create mode 100644 stovepipe/extension/queueconfig/queueconfig.go diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 7c017b50..438ced4c 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/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", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index e082ba5d..22c53a60 100644 --- a/service/stovepipe/server/main.go +++ b/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) } diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index 0299783f..a856f647 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/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", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index f66b788d..0ec2cc0b 100644 --- a/stovepipe/controller/process/process.go +++ b/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,6 +53,7 @@ func NewController( logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, + queueConfigs queueconfig.Store, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { @@ -58,6 +61,7 @@ func NewController( 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, diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 77caaa2b..2fd124e6 100644 --- a/stovepipe/controller/process/process_test.go +++ b/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", diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 104a8ec6..84220bda 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/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", ], diff --git a/stovepipe/entity/queue_config.go b/stovepipe/entity/queue_config.go new file mode 100644 index 00000000..10797ab9 --- /dev/null +++ b/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"` +} diff --git a/stovepipe/extension/queueconfig/BUILD.bazel b/stovepipe/extension/queueconfig/BUILD.bazel new file mode 100644 index 00000000..a654ec80 --- /dev/null +++ b/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"], +) diff --git a/stovepipe/extension/queueconfig/README.md b/stovepipe/extension/queueconfig/README.md new file mode 100644 index 00000000..40f88558 --- /dev/null +++ b/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. diff --git a/stovepipe/extension/queueconfig/default/BUILD.bazel b/stovepipe/extension/queueconfig/default/BUILD.bazel new file mode 100644 index 00000000..f2c8fc3d --- /dev/null +++ b/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", + ], +) diff --git a/stovepipe/extension/queueconfig/default/default.go b/stovepipe/extension/queueconfig/default/default.go new file mode 100644 index 00000000..65da07f6 --- /dev/null +++ b/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 +} diff --git a/stovepipe/extension/queueconfig/default/default_test.go b/stovepipe/extension/queueconfig/default/default_test.go new file mode 100644 index 00000000..490ff35a --- /dev/null +++ b/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) +} diff --git a/stovepipe/extension/queueconfig/mock/BUILD.bazel b/stovepipe/extension/queueconfig/mock/BUILD.bazel new file mode 100644 index 00000000..e36da38a --- /dev/null +++ b/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", + ], +) diff --git a/stovepipe/extension/queueconfig/mock/queueconfig_mock.go b/stovepipe/extension/queueconfig/mock/queueconfig_mock.go new file mode 100644 index 00000000..c0d851c3 --- /dev/null +++ b/stovepipe/extension/queueconfig/mock/queueconfig_mock.go @@ -0,0 +1,72 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: stovepipe/extension/queueconfig/queueconfig.go +// +// Generated by this command: +// +// mockgen -source=stovepipe/extension/queueconfig/queueconfig.go -destination=stovepipe/extension/queueconfig/mock/queueconfig_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockStore is a mock of Store interface. +type MockStore struct { + ctrl *gomock.Controller + recorder *MockStoreMockRecorder + isgomock struct{} +} + +// MockStoreMockRecorder is the mock recorder for MockStore. +type MockStoreMockRecorder struct { + mock *MockStore +} + +// NewMockStore creates a new mock instance. +func NewMockStore(ctrl *gomock.Controller) *MockStore { + mock := &MockStore{ctrl: ctrl} + mock.recorder = &MockStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockStore) EXPECT() *MockStoreMockRecorder { + return m.recorder +} + +// Get mocks base method. +func (m *MockStore) Get(ctx context.Context, name string) (entity.QueueConfig, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, name) + ret0, _ := ret[0].(entity.QueueConfig) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockStoreMockRecorder) Get(ctx, name any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockStore)(nil).Get), ctx, name) +} + +// List mocks base method. +func (m *MockStore) List(ctx context.Context) ([]entity.QueueConfig, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx) + ret0, _ := ret[0].([]entity.QueueConfig) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockStoreMockRecorder) List(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockStore)(nil).List), ctx) +} diff --git a/stovepipe/extension/queueconfig/queueconfig.go b/stovepipe/extension/queueconfig/queueconfig.go new file mode 100644 index 00000000..437d9a2b --- /dev/null +++ b/stovepipe/extension/queueconfig/queueconfig.go @@ -0,0 +1,38 @@ +// 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 queueconfig + +//go:generate mockgen -source=queueconfig.go -destination=mock/queueconfig_mock.go -package=mock + +import ( + "context" + "errors" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// ErrNotFound is returned when the requested queue configuration does not exist. +var ErrNotFound = errors.New("queue config not found") + +// Store loads and provides queue configurations. +// Implementations may read from YAML files, databases, remote services, etc. +type Store interface { + // Get returns the configuration for a named queue. + // Returns ErrNotFound if no configuration exists for the given name. + Get(ctx context.Context, name string) (entity.QueueConfig, error) + + // List returns all configured queues. + List(ctx context.Context) ([]entity.QueueConfig, error) +}