Skip to content

Commit 98cef3f

Browse files
mnoah1github-actions[bot]
authored andcommitted
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.
1 parent 7003035 commit 98cef3f

15 files changed

Lines changed: 344 additions & 4 deletions

File tree

service/stovepipe/server/BUILD.bazel

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ go_library(
1616
"//stovepipe/controller:go_default_library",
1717
"//stovepipe/controller/process:go_default_library",
1818
"//stovepipe/core/messagequeue:go_default_library",
19+
"//stovepipe/extension/queueconfig/default:go_default_library",
1920
"//stovepipe/extension/sourcecontrol:go_default_library",
2021
"//stovepipe/extension/sourcecontrol/fake:go_default_library",
2122
"//stovepipe/extension/storage/mysql:go_default_library",

service/stovepipe/server/main.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ import (
3838
"github.com/uber/submitqueue/stovepipe/controller"
3939
"github.com/uber/submitqueue/stovepipe/controller/process"
4040
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
41+
queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
4142
"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
4243
sourcecontrolfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake"
4344
storageMySQL "github.com/uber/submitqueue/stovepipe/extension/storage/mysql"
@@ -209,7 +210,7 @@ func run() error {
209210
),
210211
)
211212

212-
processController := process.NewController(logger.Sugar(), scope, store, stovepipemq.TopicKeyProcess, "stovepipe-process")
213+
processController := process.NewController(logger.Sugar(), scope, store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process")
213214
if err := primaryConsumer.Register(processController); err != nil {
214215
return fmt.Errorf("failed to register process controller: %w", err)
215216
}

stovepipe/controller/process/BUILD.bazel

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ go_library(
1111
"//platform/metrics:go_default_library",
1212
"//stovepipe/core/messagequeue:go_default_library",
1313
"//stovepipe/entity:go_default_library",
14+
"//stovepipe/extension/queueconfig:go_default_library",
1415
"//stovepipe/extension/storage:go_default_library",
1516
"@com_github_uber_go_tally//:go_default_library",
1617
"@org_uber_go_zap//:go_default_library",
@@ -28,6 +29,7 @@ go_test(
2829
"//platform/extension/messagequeue/mock:go_default_library",
2930
"//stovepipe/core/messagequeue:go_default_library",
3031
"//stovepipe/entity:go_default_library",
32+
"//stovepipe/extension/queueconfig/default:go_default_library",
3133
"//stovepipe/extension/storage:go_default_library",
3234
"//stovepipe/extension/storage/mock:go_default_library",
3335
"@com_github_stretchr_testify//assert:go_default_library",

stovepipe/controller/process/process.go

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import (
2929
"github.com/uber/submitqueue/platform/metrics"
3030
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
3131
"github.com/uber/submitqueue/stovepipe/entity"
32+
"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
3233
"github.com/uber/submitqueue/stovepipe/extension/storage"
3334
"go.uber.org/zap"
3435
)
@@ -39,6 +40,7 @@ type Controller struct {
3940
logger *zap.SugaredLogger
4041
metricsScope tally.Scope
4142
store storage.Storage
43+
queueConfigs queueconfig.Store
4244
topicKey consumer.TopicKey
4345
consumerGroup string
4446
}
@@ -51,13 +53,15 @@ func NewController(
5153
logger *zap.SugaredLogger,
5254
scope tally.Scope,
5355
store storage.Storage,
56+
queueConfigs queueconfig.Store,
5457
topicKey consumer.TopicKey,
5558
consumerGroup string,
5659
) *Controller {
5760
return &Controller{
5861
logger: logger.Named("process_controller"),
5962
metricsScope: scope.SubScope("process_controller"),
6063
store: store,
64+
queueConfigs: queueConfigs,
6165
topicKey: topicKey,
6266
consumerGroup: consumerGroup,
6367
}
@@ -101,8 +105,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
101105
}
102106
}
103107

104-
// processAccepted coalesces older heads against queue.latest_request_id. The latest
105-
// head is left in accepted until admit and the concurrency gate land in later PRs.
108+
// processAccepted coalesces older heads against queue.latest_request_id, then resolves
109+
// per-queue config for the concurrency gate. Admit lands in a follow-up PR.
106110
func (c *Controller) processAccepted(ctx context.Context, request entity.Request) error {
107111
queueRow, err := c.loadQueue(ctx, request.Queue)
108112
if err != nil {
@@ -134,6 +138,20 @@ func (c *Controller) processAccepted(ctx context.Context, request entity.Request
134138
return nil
135139
}
136140

141+
cfg, err := c.queueConfigs.Get(ctx, request.Queue)
142+
if err != nil {
143+
return fmt.Errorf("ProcessController failed to load queue config for %s: %w", request.Queue, err)
144+
}
145+
146+
if queueRow.InFlightCount >= cfg.MaxConcurrent {
147+
c.logger.Infow("latest head awaiting build slot",
148+
"request_id", request.ID,
149+
"queue", request.Queue,
150+
"in_flight_count", queueRow.InFlightCount,
151+
)
152+
return nil
153+
}
154+
137155
c.logger.Infow("latest head awaiting admit",
138156
"request_id", request.ID,
139157
"queue", request.Queue,

stovepipe/controller/process/process_test.go

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
2929
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
3030
"github.com/uber/submitqueue/stovepipe/entity"
31+
queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
3132
"github.com/uber/submitqueue/stovepipe/extension/storage"
3233
storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
3334
"go.uber.org/mock/gomock"
@@ -57,7 +58,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processM
5758
store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
5859
store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
5960

60-
c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, stovepipemq.TopicKeyProcess, "stovepipe-process")
61+
c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process")
6162
return c, m
6263
}
6364

@@ -131,6 +132,18 @@ func TestProcess(t *testing.T) {
131132
}, nil)
132133
},
133134
},
135+
{
136+
name: "latest accepted head awaits slot when gate closed",
137+
setup: func(m processMocks) {
138+
m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil)
139+
m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{
140+
Name: testQueue,
141+
LatestRequestID: testID,
142+
InFlightCount: 1,
143+
Version: 1,
144+
}, nil)
145+
},
146+
},
134147
{
135148
name: "older accepted head is superseded",
136149
id: "request/monorepo/main/3",

stovepipe/entity/BUILD.bazel

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ go_library(
44
name = "go_default_library",
55
srcs = [
66
"queue.go",
7+
"queue_config.go",
78
"request.go",
89
"request_id.go",
910
],

stovepipe/entity/queue_config.go

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
// Copyright (c) 2025 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package entity
16+
17+
// QueueConfig holds deployment configuration for a Stovepipe validation queue.
18+
// Mutable runtime state (latest head, in-flight count) lives on the Queue row;
19+
// knobs such as max_concurrent are resolved here at gate-check time. Immutable
20+
// after load.
21+
type QueueConfig struct {
22+
// Name uniquely identifies this queue within the system.
23+
// Referenced by Request.Queue.
24+
Name string `json:"name" yaml:"name"`
25+
// MaxConcurrent is the cap on concurrent in-flight validations for the queue.
26+
MaxConcurrent int32 `json:"max_concurrent" yaml:"max_concurrent"`
27+
// GateWaitDelayMs is the PublishAfter delay when the latest head waits for a slot.
28+
GateWaitDelayMs int64 `json:"gate_wait_delay_ms" yaml:"gate_wait_delay_ms"`
29+
}
Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
load("@rules_go//go:def.bzl", "go_library")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["queueconfig.go"],
6+
importpath = "github.com/uber/submitqueue/stovepipe/extension/queueconfig",
7+
visibility = ["//visibility:public"],
8+
deps = ["//stovepipe/entity:go_default_library"],
9+
)
Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
# Queue Config Extension
2+
3+
Vendor-agnostic interface for providing Stovepipe queue configurations.
4+
5+
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).
6+
7+
## Interfaces
8+
9+
`Store` provides queue configurations by name via `Get` and `List`. See `queueconfig.go`.
10+
11+
## Entities
12+
13+
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.
14+
15+
## Implementations
16+
17+
`default` returns the global wiring defaults for any non-empty queue name until a file- or service-backed store lands.
Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["default.go"],
6+
importpath = "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//stovepipe/entity:go_default_library",
10+
"//stovepipe/extension/queueconfig:go_default_library",
11+
],
12+
)
13+
14+
go_test(
15+
name = "go_default_test",
16+
srcs = ["default_test.go"],
17+
embed = [":go_default_library"],
18+
deps = [
19+
"//stovepipe/extension/queueconfig:go_default_library",
20+
"@com_github_stretchr_testify//assert:go_default_library",
21+
"@com_github_stretchr_testify//require:go_default_library",
22+
],
23+
)

0 commit comments

Comments
 (0)