From c6cb600a6e810a6eaff6b19b4b94101f1bd4bb34 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 15 Sep 2026 20:56:47 +0000 Subject: [PATCH 1/5] feat(stovepipe): configure admission policies per queue Summary: Intent: - Make logical admission throttling configurable per Stovepipe queue. Changes: - Add a strict, validated YAML-backed queue configuration store. - Wire optional QUEUE_CONFIG_PATH loading into the Stovepipe server and local stack. - Validate configured queues against message-queue tenants and document duration settings. This change builds on the general admission throttle introduced by the parent PR. --- doc/rfc/stovepipe/steps/process.md | 2 +- service/stovepipe/README.md | 3 + service/stovepipe/docker-compose.yml | 1 + service/stovepipe/server/BUILD.bazel | 4 + service/stovepipe/server/Dockerfile | 1 + service/stovepipe/server/main.go | 42 +++++- service/stovepipe/server/main_test.go | 41 +++++- service/stovepipe/server/queues.yaml | 15 +++ stovepipe/extension/queueconfig/README.md | 5 +- .../extension/queueconfig/yaml/BUILD.bazel | 25 ++++ stovepipe/extension/queueconfig/yaml/yaml.go | 96 ++++++++++++++ .../extension/queueconfig/yaml/yaml_test.go | 122 ++++++++++++++++++ 12 files changed, 352 insertions(+), 5 deletions(-) create mode 100644 service/stovepipe/server/queues.yaml create mode 100644 stovepipe/extension/queueconfig/yaml/BUILD.bazel create mode 100644 stovepipe/extension/queueconfig/yaml/yaml.go create mode 100644 stovepipe/extension/queueconfig/yaml/yaml_test.go diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index 8f751d9af..80aa719c2 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/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. diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index 78668eef2..9c67f31a9 100644 --- a/service/stovepipe/README.md +++ b/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-` | diff --git a/service/stovepipe/docker-compose.yml b/service/stovepipe/docker-compose.yml index 91b5e0e0d..f476fa7d9 100644 --- a/service/stovepipe/docker-compose.yml +++ b/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: diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index bd8fa24ac..868c17261 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/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", diff --git a/service/stovepipe/server/Dockerfile b/service/stovepipe/server/Dockerfile index adbe1aebb..c5dff5bb1 100644 --- a/service/stovepipe/server/Dockerfile +++ b/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 diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index ce5d2dcdf..f4a85f348 100644 --- a/service/stovepipe/server/main.go +++ b/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( diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index e8f1d6061..d7c15c37c 100644 --- a/service/stovepipe/server/main_test.go +++ b/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) + }) +} diff --git a/service/stovepipe/server/queues.yaml b/service/stovepipe/server/queues.yaml new file mode 100644 index 000000000..07d70d20c --- /dev/null +++ b/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 diff --git a/stovepipe/extension/queueconfig/README.md b/stovepipe/extension/queueconfig/README.md index 74708b661..9ac820909 100644 --- a/stovepipe/extension/queueconfig/README.md +++ b/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. diff --git a/stovepipe/extension/queueconfig/yaml/BUILD.bazel b/stovepipe/extension/queueconfig/yaml/BUILD.bazel new file mode 100644 index 000000000..ffd790e01 --- /dev/null +++ b/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", + ], +) diff --git a/stovepipe/extension/queueconfig/yaml/yaml.go b/stovepipe/extension/queueconfig/yaml/yaml.go new file mode 100644 index 000000000..8689086b4 --- /dev/null +++ b/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 +} diff --git a/stovepipe/extension/queueconfig/yaml/yaml_test.go b/stovepipe/extension/queueconfig/yaml/yaml_test.go new file mode 100644 index 000000000..f4dc27d0e --- /dev/null +++ b/stovepipe/extension/queueconfig/yaml/yaml_test.go @@ -0,0 +1,122 @@ +// 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 + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/queueconfig" +) + +func writeQueueConfig(t *testing.T, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "queues.yaml") + require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) + return path +} + +func TestNewStore(t *testing.T) { + tests := []struct { + name string + content string + wantErr bool + }{ + { + name: "loads admission policies", + content: `queues: + - name: monorepo/main + max_concurrent: 2 + gate_wait_delay_ms: 5000 + minimum_build_admission_interval_ms: 3600000 +`, + }, + {name: "empty list", content: "queues: []\n"}, + {name: "empty document", content: "", wantErr: true}, + {name: "malformed YAML", content: "queues: [", wantErr: true}, + {name: "unknown field", content: validQueueYAML("main", 1, 5000, 0) + "unexpected: true\n", wantErr: true}, + {name: "empty name", content: validQueueYAML("", 1, 5000, 0), wantErr: true}, + {name: "non-positive concurrency", content: validQueueYAML("main", 0, 5000, 0), wantErr: true}, + {name: "non-positive gate delay", content: validQueueYAML("main", 1, 0, 0), wantErr: true}, + {name: "negative minimum interval disables policy", content: validQueueYAML("main", 1, 5000, -1)}, + { + name: "duplicate name", + content: validQueueYAML("main", 1, 5000, 0) + ` - name: main + max_concurrent: 1 + gate_wait_delay_ms: 5000 +`, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store, err := NewStore(writeQueueConfig(t, tt.content)) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + _, err = store.List(context.Background()) + require.NoError(t, err) + }) + } +} + +func validQueueYAML(name string, maxConcurrent int32, gateWaitDelayMs, minimumIntervalMs int64) string { + return fmt.Sprintf(`queues: + - name: %s + max_concurrent: %d + gate_wait_delay_ms: %d + minimum_build_admission_interval_ms: %d +`, name, maxConcurrent, gateWaitDelayMs, minimumIntervalMs) +} + +func TestStoreGetAndList(t *testing.T) { + store, err := NewStore(writeQueueConfig(t, validQueueYAML("monorepo/main", 2, 5000, 3_600_000))) + require.NoError(t, err) + + got, err := store.Get(context.Background(), "monorepo/main") + require.NoError(t, err) + assert.Equal(t, entity.QueueConfig{ + Name: "monorepo/main", + MaxConcurrent: 2, + GateWaitDelayMs: 5000, + MinimumBuildAdmissionIntervalMs: 3_600_000, + }, got) + + _, err = store.Get(context.Background(), "missing") + assert.ErrorIs(t, err, queueconfig.ErrNotFound) + + first, err := store.List(context.Background()) + require.NoError(t, err) + first[0].Name = "changed" + second, err := store.List(context.Background()) + require.NoError(t, err) + assert.Equal(t, "monorepo/main", second[0].Name) +} + +func TestNewStoreMissingFile(t *testing.T) { + _, err := NewStore(filepath.Join(t.TempDir(), "missing.yaml")) + require.Error(t, err) + assert.False(t, errors.Is(err, queueconfig.ErrNotFound)) +} From 9a1946d71de3307dcbe199055ff969ef422ffdec Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 15 Sep 2026 21:25:30 +0000 Subject: [PATCH 2/5] fix(stovepipe): stage queue config in compose tests --- test/e2e/stovepipe/suite_test.go | 5 +++-- test/integration/stovepipe/suite_test.go | 5 +++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index c8084e408..c53fe5340 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -86,8 +86,9 @@ func (s *StovepipeE2ESuite) SetupSuite() { composeFile := testutil.Runfile("service/stovepipe/docker-compose.yml") s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "e2e-stovepipe", testutil.WithBuildContext(map[string]string{ - ".docker-bin/stovepipe": "service/stovepipe/server/stovepipe_linux", - "service/stovepipe/server/Dockerfile": "service/stovepipe/server/Dockerfile", + ".docker-bin/stovepipe": "service/stovepipe/server/stovepipe_linux", + "service/stovepipe/server/Dockerfile": "service/stovepipe/server/Dockerfile", + "service/stovepipe/server/queues.yaml": "service/stovepipe/server/queues.yaml", })) err := s.stack.Up() diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index 1a891783f..dd21919d7 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/test/integration/stovepipe/suite_test.go @@ -68,8 +68,9 @@ func (s *StovepipeIntegrationSuite) SetupSuite() { composeFile := testutil.Runfile("service/stovepipe/docker-compose.yml") s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "svc-stovepipe", testutil.WithBuildContext(map[string]string{ - ".docker-bin/stovepipe": "service/stovepipe/server/stovepipe_linux", - "service/stovepipe/server/Dockerfile": "service/stovepipe/server/Dockerfile", + ".docker-bin/stovepipe": "service/stovepipe/server/stovepipe_linux", + "service/stovepipe/server/Dockerfile": "service/stovepipe/server/Dockerfile", + "service/stovepipe/server/queues.yaml": "service/stovepipe/server/queues.yaml", })) err := s.stack.Up() From 9c411aa1558ced13a400923a523597d13ace3038 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Thu, 17 Sep 2026 19:56:56 +0000 Subject: [PATCH 3/5] feat(stovepipe): define admission gate framework --- Makefile | 2 +- doc/rfc/index.md | 1 + doc/rfc/stovepipe/admission-gate.md | 190 ++++++++++++++++++ stovepipe/extension/admissiongate/BUILD.bazel | 9 + stovepipe/extension/admissiongate/README.md | 5 + .../extension/admissiongate/admissiongate.go | 80 ++++++++ .../extension/admissiongate/mock/BUILD.bazel | 13 ++ .../admissiongate/mock/admissiongate_mock.go | 97 +++++++++ 8 files changed, 396 insertions(+), 1 deletion(-) create mode 100644 doc/rfc/stovepipe/admission-gate.md create mode 100644 stovepipe/extension/admissiongate/BUILD.bazel create mode 100644 stovepipe/extension/admissiongate/README.md create mode 100644 stovepipe/extension/admissiongate/admissiongate.go create mode 100644 stovepipe/extension/admissiongate/mock/BUILD.bazel create mode 100644 stovepipe/extension/admissiongate/mock/admissiongate_mock.go diff --git a/Makefile b/Makefile index b15be5161..448da31b8 100644 --- a/Makefile +++ b/Makefile @@ -579,7 +579,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service mocks: ## Generate mock files using mockgen @echo "Generating mocks..." - @$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... + @$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/admissiongate/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... @echo "Mocks generated successfully!" proto: ## Generate protobuf files from .proto definitions diff --git a/doc/rfc/index.md b/doc/rfc/index.md index 11dcb60d9..2a23a60c4 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -27,6 +27,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting ## Stovepipe - [Stovepipe Workflow](stovepipe/workflow.md) - Post-land validation pipeline overview: ingest, process, build, record greenness, analyze projects, notify downstream +- [Admission Gates](stovepipe/admission-gate.md) - Extensible, queue-scoped logical admission decisions with atomic policy composition, opaque versioned state, optimistic locking, and reconciliation from durable request outcomes - [Process stage](stovepipe/steps/process.md) - Build-strategy decision, per-queue concurrency gate, backlog coalescing, entity model, platform prerequisites - [Build stage](stovepipe/steps/build.md) - Trigger-only stage and Stovepipe's URI-based BuildRunner contract - [Buildsignal stage](stovepipe/steps/buildsignal.md) - Build polling, terminal status persistence, and the handoff to record diff --git a/doc/rfc/stovepipe/admission-gate.md b/doc/rfc/stovepipe/admission-gate.md new file mode 100644 index 000000000..28b74647d --- /dev/null +++ b/doc/rfc/stovepipe/admission-gate.md @@ -0,0 +1,190 @@ +# Stovepipe Admission Gates + +Status: proposed. This RFC defines the framework and extension contract; storage and controller integration land separately. + +## Problem + +Stovepipe currently makes its build-admission decisions directly in `process`: admit only below the queue's concurrency limit, and optionally delay starts by a minimum interval. A failure cooldown adds a third decision with the same shape, but implementing each rule in a controller spreads policy across lifecycle stages and makes every new rule another special case. + +The immediate requirements are: + +- Limit concurrent logical validations per queue. +- Throttle admissions generally, for example to at most one start per hour. +- After a build runner reports a failed result, defer the next admission for a configured cooldown. +- Keep coalescing active while a request is deferred, so a newer head can supersede it without waiting for the gate to open. + +The framework must also leave room for policies such as maintenance windows, resource budgets, provider health, or an operator hold, and for logical admission points other than build admission. + +## Scope + +An **admission gate** decides whether one domain entity may cross a logical pipeline boundary. The first point is `build`: whether the latest accepted Stovepipe request may reserve validation capacity and advance toward the build stage. + +This is separate from the shared [Consumer Gate](../consumer-gate.md). A consumer gate is an external operational control that stops deliveries before a controller. A Stovepipe admission gate is domain policy evaluated by a controller for a specific request. Both defer with queue redelivery, but they answer different questions and own different state. + +## Contract + +The vendor-neutral contract lives at `stovepipe/extension/admissiongate`: + +```go +type Point string + +const PointBuild Point = "build" + +type Result struct { + Decision Decision + BlockedBy []string +} + +type Gate interface { + TryAdmit(context.Context, entity.Request) (Result, error) +} + +type Factory interface { + For(Config) (Gate, error) +} +``` + +`Point` is an open string identifier. The shared package names only points understood by Stovepipe; adding a point is additive and does not change `Gate`. `Config` carries the queue and point so host wiring can select an implementation without putting routing in an extension package. + +`TryAdmit` takes the thin `entity.Request`, following the repository's identity-in extension rule. The request already identifies its queue. A gate resolves the queue-scoped storage, configuration, request history, clocks, or remote services it needs through dependencies injected when its implementation is constructed. Controllers do not pre-resolve policy facts and hand them across the contract. + +`Result` represents expected control flow: + +- `DecisionAdmitted` means this request's admission was already recorded or has been durably recorded before the call returns. +- `DecisionDeferred` means no admission was recorded because one or more policies currently block it. `BlockedBy` contains stable, low-cardinality policy identifiers for logs and metrics, not human-readable errors. +- `DecisionUnknown` is the invalid zero value and must be treated as an implementation failure. + +Errors are reserved for failures to evaluate or durably record the decision. A closed gate is not an error and does not consume retry budget. + +There is intentionally no `Complete`, `Release`, or `RecordOutcome` method. `TryAdmit` must be able to derive the current answer from durable state. This keeps build outcomes owned by the request lifecycle and prevents `buildsignal`, DLQ controllers, or future terminal paths from each needing policy-specific callbacks. + +## Process Integration + +`process` retains responsibility for request choreography; the gate owns only admission policy and its reservation: + +1. Load the request and queue, then coalesce it against the latest request ID. +2. Resolve the gate for `(request.Queue, PointBuild)`. +3. Call `TryAdmit(ctx, request)`. +4. On `DecisionDeferred`, hold the delivery for the queue's normal gate re-check delay and return successfully. +5. On redelivery, start again at coalescing before evaluating the gate. +6. On `DecisionAdmitted`, derive the build strategy, transition the request to `processing`, and publish to `build` using the existing persist-before-publish ordering. + +The hold delay belongs to controller scheduling configuration, not `Result`. A policy may know an exact deadline, but sleeping until that deadline would suppress coalescing for its full duration. Frequent bounded re-checks preserve superseding and make all policies converge through the same path. + +If the process dies after the gate records admission but before the request reaches `processing`, redelivery calls `TryAdmit` with the same request ID. The result is admitted without reserving twice, and the controller retries the transition. If the process message ultimately reaches its DLQ, the DLQ's terminal request transition becomes visible to later reconciliation. + +## Durable State + +The first implementation adds `AdmissionState []byte` to `entity.Queue` and appends an `admission_state BLOB` column to the end of the MySQL queue schema. `QueueStore` only round-trips those bytes as part of the existing versioned queue snapshot; it does not parse, validate, merge, or version the payload. + +The concrete gate exclusively owns the payload's encoding and compatibility. The initial implementation uses versioned JSON because the state is small and operationally inspectable, but the storage contract is opaque bytes rather than a JSON contract. A different implementation may use protobuf or another encoding. Changing the implementation for a live queue requires that the replacement understand or explicitly migrate the prior payload. + +A representative initial payload is: + +```json +{ + "version": 1, + "last_admitted_request_id": "request/monorepo/main/42", + "active": { + "request/monorepo/main/42": { + "admitted_at_ms": 1789506000000 + } + }, + "policies": { + "minimum_interval": { + "last_admitted_at_ms": 1789506000000 + }, + "failure_cooldown": { + "not_before_ms": 1789509600000, + "source_request_id": "request/monorepo/main/41" + } + } +} +``` + +This shape illustrates ownership, not a shared wire contract. Policy keys and values are namespaced inside the implementation's versioned envelope. Unrelated controllers never mutate individual keys, and independently selected extensions never share a metadata map. + +Storing the envelope on the queue gives admission one optimistic-lock boundary with the queue's latest-head pointer and other coordination fields. The gate loads the complete queue snapshot, changes only its owned field, computes `newVersion = oldVersion + 1`, performs the conditional write, and assigns the new version only after success. On `ErrVersionMismatch`, it reloads and restarts evaluation. It preserves concurrent changes to fields it does not own by always rebuilding from the reloaded snapshot. + +No cross-entity transaction is introduced. Recording an admission and transitioning its request to `processing` remain two convergent versioned writes. The request ID is the idempotency key joining them. + +## Independent Reconciliation + +The state retains the IDs and admission times of active requests. Before evaluating a new admission, the gate reconciles that bounded set against authoritative request storage: + +- `accepted` or `processing` remains active. +- A terminal request is removed from the active set. +- Its terminal request-log record supplies the outcome reason and occurrence time needed by outcome-sensitive policies. +- If the request is terminal but its corresponding log is not visible yet, evaluation defers. The lifecycle writer will retry the log write, and a later gate evaluation converges. + +The lookup cost is bounded by the configured concurrency limit rather than queue history size. It uses primary-key request reads and request-owned log reads; it requires no query by status or new secondary index. + +Using request history is what lets failure cooldown mean "after the runner-reported failure" rather than "after some later admission attempt noticed a failure." The initial cooldown policy reacts only to `RequestOutcomeReasonBuildFailed`. Success, cancellation, superseding, and failures synthesized by a DLQ or timeout do not activate it unless a later policy explicitly chooses those reasons. + +`buildsignal` therefore remains policy-neutral. It persists the build result, request terminal state, and request log. It neither understands the gate envelope nor invokes an admission callback. + +## Policy Composition + +One resolved `Gate` is the atomic composition boundary for one queue and admission point. The implementation may contain several policies, but they are not independently stateful extensions called in sequence. + +For each `TryAdmit`, the implementation: + +1. Loads and decodes one state snapshot. +2. Reconciles completed admissions into policy facts. +3. Evaluates every enabled policy against the same snapshot. +4. If any policy blocks, persists reconciliation changes if needed but records no new admission. +5. If all policies allow, applies every policy's admission mutation and the active-request reservation to one new snapshot and commits it with one queue CAS. + +This prevents partial admission: an interval policy cannot consume its next slot only for a later budget policy to reject the request. Policy evaluation and proposed mutations may be separate internal primitives in the standard implementation, but they are deliberately absent from the public extension contract. That leaves alternative gate implementations free to use a remote quota service, a rules engine, or a single purpose-built algorithm without emulating an in-process policy interface. + +The initial standard gate composes: + +- **Concurrency:** defer while the number of reconciled active admissions is at the per-queue limit. +- **Minimum interval:** when configured above zero, require at least that many milliseconds between admission timestamps. Non-positive values disable it. +- **Failure cooldown:** after a configured runner-reported failure, defer until the failure occurrence time plus the cooldown. Non-positive values disable it. + +Future policies at the same point, such as a calendar window, cost budget, provider-health circuit, or manual hold, fit inside the same atomic composition. A policy that needs its own durable facts receives a namespaced section in the gate envelope. A fundamentally different backend or evaluation model is another `Gate` implementation selected by wiring. + +## Configuration And Routing + +Queue policy settings remain deployment configuration supplied through `queueconfig`; mutable observations remain in `AdmissionState`. Configuration is read during evaluation so a changed interval or cooldown affects the next attempt without rewriting stored state. + +The common admission-gate contract does not define a universal policy configuration language. The standard gate understands Stovepipe's typed queue settings. Another gate may receive configuration through dependencies injected by its constructor. Per-queue and per-point selection belongs in service wiring through `Factory.For(Config)`, consistent with other extensions; no implementation package contains a routing map. + +This separation permits gradual evolution. The current build point can use the standard composite gate, a high-cost queue can use a remote budget gate, and a future promotion point can use an independently selected gate without changing controllers' result handling. + +## Observability + +The controller records admitted and deferred counters tagged by admission point. Deferred counters may additionally use each `BlockedBy` identifier, whose low-cardinality contract makes it safe as a metric tag. Logs include request ID, queue, point, decision, and blockers. + +The gate implementation records evaluation, state decode, reconciliation, CAS-conflict, and dependency errors. It must not place opaque state contents or arbitrary configuration values in metric tags. + +An operator can inspect the initial JSON state in MySQL, but that is diagnostic only. No controller, API, or operational tool may depend on its internal keys without going through an implementation-owned decoder. + +## Failure Posture + +Admission gates fail closed. An error loading configuration, resolving storage, decoding state, reconciling outcomes, or committing an admission returns an error from `TryAdmit`; the controller does not advance the request. Normal consumer retry and DLQ behavior handles persistent infrastructure or configuration failures. + +Unknown state versions also fail closed. Silently resetting an unreadable payload could over-admit work or discard a live cooldown. Rollouts that change the codec must support mixed-version readers and writers for the deployment window or migrate queues before switching implementations. + +## Rollout + +The implementation PR can migrate incrementally: + +1. Add the opaque queue field and append the MySQL column, treating empty bytes as version-1 empty state. +2. Implement the standard composite gate with concurrency and minimum-interval policies matching current behavior. +3. Wire `process` through `Factory` and remove its direct admission-counter/deadline decisions. +4. Stop `buildsignal` from directly releasing admission capacity; reconciliation becomes authoritative. +5. Add failure cooldown as another standard policy. + +During a rolling deployment, old and new processes must not concurrently own different representations of admission capacity. The wiring cutover therefore occurs only after every binary understands the new queue field, or behind a deployment-wide switch that keeps one ownership model active at a time. + +## Rejected + +- **A callback from every terminal path.** A `Complete` or `RecordOutcome` method makes correctness depend on buildsignal, cancellation, timeout, and DLQ paths all notifying the gate exactly once. Reconciliation from durable request facts is simpler and converges after missed work. +- **One extension call per policy.** Stateful gates called sequentially cannot atomically roll back earlier reservations when a later policy blocks. One gate owns composition and one CAS boundary. +- **Policy fields as queue columns.** Typed columns are easy for the first interval and cooldown but require schema and storage-contract changes for every new policy. An opaque, ownership-specific field keeps storage backend-neutral. +- **A generic queue metadata map.** It obscures ownership and invites unrelated controllers to mutate extension-defined keys. `admission_state` names one writer and one compatibility contract. +- **Sleeping until a policy deadline.** A long hold delays the next coalescing check. The normal short deferred wait keeps superseding responsive. +- **In-memory reservations.** They diverge across replicas and disappear on restart. Durable queue-scoped state plus optimistic locking is required for distributed admission. +- **Failing open on gate errors.** Admission exists to protect finite or costly resources; an unavailable policy dependency must not silently remove that protection. diff --git a/stovepipe/extension/admissiongate/BUILD.bazel b/stovepipe/extension/admissiongate/BUILD.bazel new file mode 100644 index 000000000..9a28cf3b4 --- /dev/null +++ b/stovepipe/extension/admissiongate/BUILD.bazel @@ -0,0 +1,9 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["admissiongate.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/admissiongate", + visibility = ["//visibility:public"], + deps = ["//stovepipe/entity:go_default_library"], +) diff --git a/stovepipe/extension/admissiongate/README.md b/stovepipe/extension/admissiongate/README.md new file mode 100644 index 000000000..5917b444f --- /dev/null +++ b/stovepipe/extension/admissiongate/README.md @@ -0,0 +1,5 @@ +# Admission Gate Extension + +Vendor-neutral contract for deciding whether a Stovepipe request may cross a logical admission point. See the [Stovepipe Admission Gates RFC](../../../doc/rfc/stovepipe/admission-gate.md) for decision semantics, durable-state ownership, composition, and controller integration. + +Implementations take request identity, resolve their own facts, and return an admitted or deferred result. Per-queue and per-point implementation routing belongs in service wiring. diff --git a/stovepipe/extension/admissiongate/admissiongate.go b/stovepipe/extension/admissiongate/admissiongate.go new file mode 100644 index 000000000..a1a77d81d --- /dev/null +++ b/stovepipe/extension/admissiongate/admissiongate.go @@ -0,0 +1,80 @@ +// 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 admissiongate defines Stovepipe's extension contract for deciding +// whether a request may cross a logical pipeline admission point. +package admissiongate + +//go:generate mockgen -source=admissiongate.go -destination=mock/admissiongate_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// Point identifies a logical pipeline boundary guarded by an admission gate. +type Point string + +const ( + // PointUnknown is the invalid zero value. + PointUnknown Point = "" + // PointBuild guards admission of a request toward the build stage. + PointBuild Point = "build" +) + +// Decision is the expected outcome of evaluating an admission gate. +type Decision string + +const ( + // DecisionUnknown is the invalid zero value. + DecisionUnknown Decision = "" + // DecisionAdmitted means the request's admission is durably reserved. + DecisionAdmitted Decision = "admitted" + // DecisionDeferred means policy currently prevents admission. + DecisionDeferred Decision = "deferred" +) + +// Result describes an expected admission outcome. +type Result struct { + // Decision is whether the request was admitted or deferred. + Decision Decision + // BlockedBy contains stable policy identifiers when Decision is deferred. + BlockedBy []string +} + +// Gate decides whether requests may cross one queue-scoped admission point. +// Implementations resolve the durable facts they need from the request's +// identity and must make repeated calls for the same request idempotent. +type Gate interface { + // TryAdmit evaluates current policy and durably reserves an allowed + // admission before returning DecisionAdmitted. A policy denial returns + // DecisionDeferred rather than an error. + TryAdmit(ctx context.Context, request entity.Request) (Result, error) +} + +// Config identifies the queue and admission point resolved by a Factory. +type Config struct { + // QueueName identifies the queue whose requests the Gate evaluates. + QueueName string + // Point identifies the logical pipeline boundary the Gate protects. + Point Point +} + +// Factory resolves queue- and point-scoped admission gates. Concrete routing +// belongs in service wiring rather than an extension implementation package. +type Factory interface { + // For returns the Gate selected for cfg. + For(cfg Config) (Gate, error) +} diff --git a/stovepipe/extension/admissiongate/mock/BUILD.bazel b/stovepipe/extension/admissiongate/mock/BUILD.bazel new file mode 100644 index 000000000..9ca4569b0 --- /dev/null +++ b/stovepipe/extension/admissiongate/mock/BUILD.bazel @@ -0,0 +1,13 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["admissiongate_mock.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/admissiongate/mock", + visibility = ["//visibility:public"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/admissiongate:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go b/stovepipe/extension/admissiongate/mock/admissiongate_mock.go new file mode 100644 index 000000000..4313ad159 --- /dev/null +++ b/stovepipe/extension/admissiongate/mock/admissiongate_mock.go @@ -0,0 +1,97 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: admissiongate.go +// +// Generated by this command: +// +// mockgen -source=admissiongate.go -destination=mock/admissiongate_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" + admissiongate "github.com/uber/submitqueue/stovepipe/extension/admissiongate" + gomock "go.uber.org/mock/gomock" +) + +// MockGate is a mock of Gate interface. +type MockGate struct { + ctrl *gomock.Controller + recorder *MockGateMockRecorder + isgomock struct{} +} + +// MockGateMockRecorder is the mock recorder for MockGate. +type MockGateMockRecorder struct { + mock *MockGate +} + +// NewMockGate creates a new mock instance. +func NewMockGate(ctrl *gomock.Controller) *MockGate { + mock := &MockGate{ctrl: ctrl} + mock.recorder = &MockGateMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockGate) EXPECT() *MockGateMockRecorder { + return m.recorder +} + +// TryAdmit mocks base method. +func (m *MockGate) TryAdmit(ctx context.Context, request entity.Request) (admissiongate.Result, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "TryAdmit", ctx, request) + ret0, _ := ret[0].(admissiongate.Result) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// TryAdmit indicates an expected call of TryAdmit. +func (mr *MockGateMockRecorder) TryAdmit(ctx, request any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TryAdmit", reflect.TypeOf((*MockGate)(nil).TryAdmit), ctx, request) +} + +// MockFactory is a mock of Factory interface. +type MockFactory struct { + ctrl *gomock.Controller + recorder *MockFactoryMockRecorder + isgomock struct{} +} + +// MockFactoryMockRecorder is the mock recorder for MockFactory. +type MockFactoryMockRecorder struct { + mock *MockFactory +} + +// NewMockFactory creates a new mock instance. +func NewMockFactory(ctrl *gomock.Controller) *MockFactory { + mock := &MockFactory{ctrl: ctrl} + mock.recorder = &MockFactoryMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockFactory) For(cfg admissiongate.Config) (admissiongate.Gate, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", cfg) + ret0, _ := ret[0].(admissiongate.Gate) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// For indicates an expected call of For. +func (mr *MockFactoryMockRecorder) For(cfg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), cfg) +} From e19ad4f7f8d2e57701876b0b251756bc877158e8 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Thu, 17 Sep 2026 20:24:23 +0000 Subject: [PATCH 4/5] refactor(stovepipe): resolve admission gates by request --- doc/rfc/stovepipe/admission-gate.md | 14 +++++---- stovepipe/extension/admissiongate/README.md | 2 +- .../extension/admissiongate/admissiongate.go | 19 ++++-------- .../admissiongate/mock/admissiongate_mock.go | 30 +++++++++---------- 4 files changed, 30 insertions(+), 35 deletions(-) diff --git a/doc/rfc/stovepipe/admission-gate.md b/doc/rfc/stovepipe/admission-gate.md index 28b74647d..ade871f77 100644 --- a/doc/rfc/stovepipe/admission-gate.md +++ b/doc/rfc/stovepipe/admission-gate.md @@ -39,12 +39,14 @@ type Gate interface { TryAdmit(context.Context, entity.Request) (Result, error) } -type Factory interface { - For(Config) (Gate, error) +type Gates interface { + For(Point, entity.Request) (Gate, error) } ``` -`Point` is an open string identifier. The shared package names only points understood by Stovepipe; adding a point is additive and does not change `Gate`. `Config` carries the queue and point so host wiring can select an implementation without putting routing in an extension package. +`Point` is an open string identifier. The shared package names only points understood by Stovepipe; adding a point is additive and does not change `Gate`. + +`Gates` is the host-owned resolver across queues and points. `For` takes the request rather than a separate queue configuration because the request already carries its authoritative queue identity. It returns exactly one composite gate: returning a slice of independently stateful gates would make atomic admission impossible when one gate records a reservation before a later gate defers. Concrete routing belongs in service wiring, not an extension implementation package. `TryAdmit` takes the thin `entity.Request`, following the repository's identity-in extension rule. The request already identifies its queue. A gate resolves the queue-scoped storage, configuration, request history, clocks, or remote services it needs through dependencies injected when its implementation is constructed. Controllers do not pre-resolve policy facts and hand them across the contract. @@ -63,7 +65,7 @@ There is intentionally no `Complete`, `Release`, or `RecordOutcome` method. `Try `process` retains responsibility for request choreography; the gate owns only admission policy and its reservation: 1. Load the request and queue, then coalesce it against the latest request ID. -2. Resolve the gate for `(request.Queue, PointBuild)`. +2. Resolve the gate with `gates.For(PointBuild, request)`. 3. Call `TryAdmit(ctx, request)`. 4. On `DecisionDeferred`, hold the delivery for the queue's normal gate re-check delay and return successfully. 5. On redelivery, start again at coalescing before evaluating the gate. @@ -149,7 +151,7 @@ Future policies at the same point, such as a calendar window, cost budget, provi Queue policy settings remain deployment configuration supplied through `queueconfig`; mutable observations remain in `AdmissionState`. Configuration is read during evaluation so a changed interval or cooldown affects the next attempt without rewriting stored state. -The common admission-gate contract does not define a universal policy configuration language. The standard gate understands Stovepipe's typed queue settings. Another gate may receive configuration through dependencies injected by its constructor. Per-queue and per-point selection belongs in service wiring through `Factory.For(Config)`, consistent with other extensions; no implementation package contains a routing map. +The common admission-gate contract does not define a universal policy configuration language. The standard gate understands Stovepipe's typed queue settings. Another gate may receive configuration through dependencies injected by its constructor. Per-queue and per-point selection belongs in service wiring through `Gates.For`, consistent with other plural resolver contracts; no implementation package contains a routing map. This separation permits gradual evolution. The current build point can use the standard composite gate, a high-cost queue can use a remote budget gate, and a future promotion point can use an independently selected gate without changing controllers' result handling. @@ -173,7 +175,7 @@ The implementation PR can migrate incrementally: 1. Add the opaque queue field and append the MySQL column, treating empty bytes as version-1 empty state. 2. Implement the standard composite gate with concurrency and minimum-interval policies matching current behavior. -3. Wire `process` through `Factory` and remove its direct admission-counter/deadline decisions. +3. Wire `process` through `Gates` and remove its direct admission-counter/deadline decisions. 4. Stop `buildsignal` from directly releasing admission capacity; reconciliation becomes authoritative. 5. Add failure cooldown as another standard policy. diff --git a/stovepipe/extension/admissiongate/README.md b/stovepipe/extension/admissiongate/README.md index 5917b444f..c334f540a 100644 --- a/stovepipe/extension/admissiongate/README.md +++ b/stovepipe/extension/admissiongate/README.md @@ -2,4 +2,4 @@ Vendor-neutral contract for deciding whether a Stovepipe request may cross a logical admission point. See the [Stovepipe Admission Gates RFC](../../../doc/rfc/stovepipe/admission-gate.md) for decision semantics, durable-state ownership, composition, and controller integration. -Implementations take request identity, resolve their own facts, and return an admitted or deferred result. Per-queue and per-point implementation routing belongs in service wiring. +Implementations take request identity, resolve their own facts, and return an admitted or deferred result. The `Gates` resolver selects one composite gate by request and admission point; its concrete routing belongs in service wiring. diff --git a/stovepipe/extension/admissiongate/admissiongate.go b/stovepipe/extension/admissiongate/admissiongate.go index a1a77d81d..afcf80b4f 100644 --- a/stovepipe/extension/admissiongate/admissiongate.go +++ b/stovepipe/extension/admissiongate/admissiongate.go @@ -64,17 +64,10 @@ type Gate interface { TryAdmit(ctx context.Context, request entity.Request) (Result, error) } -// Config identifies the queue and admission point resolved by a Factory. -type Config struct { - // QueueName identifies the queue whose requests the Gate evaluates. - QueueName string - // Point identifies the logical pipeline boundary the Gate protects. - Point Point -} - -// Factory resolves queue- and point-scoped admission gates. Concrete routing -// belongs in service wiring rather than an extension implementation package. -type Factory interface { - // For returns the Gate selected for cfg. - For(cfg Config) (Gate, error) +// Gates resolves the gate for a request and logical admission point. Concrete +// routing belongs in service wiring rather than an extension implementation +// package. +type Gates interface { + // For returns the Gate selected for point and request. + For(point Point, request entity.Request) (Gate, error) } diff --git a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go b/stovepipe/extension/admissiongate/mock/admissiongate_mock.go index 4313ad159..ede85e46a 100644 --- a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go +++ b/stovepipe/extension/admissiongate/mock/admissiongate_mock.go @@ -57,41 +57,41 @@ func (mr *MockGateMockRecorder) TryAdmit(ctx, request any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TryAdmit", reflect.TypeOf((*MockGate)(nil).TryAdmit), ctx, request) } -// MockFactory is a mock of Factory interface. -type MockFactory struct { +// MockGates is a mock of Gates interface. +type MockGates struct { ctrl *gomock.Controller - recorder *MockFactoryMockRecorder + recorder *MockGatesMockRecorder isgomock struct{} } -// MockFactoryMockRecorder is the mock recorder for MockFactory. -type MockFactoryMockRecorder struct { - mock *MockFactory +// MockGatesMockRecorder is the mock recorder for MockGates. +type MockGatesMockRecorder struct { + mock *MockGates } -// NewMockFactory creates a new mock instance. -func NewMockFactory(ctrl *gomock.Controller) *MockFactory { - mock := &MockFactory{ctrl: ctrl} - mock.recorder = &MockFactoryMockRecorder{mock} +// NewMockGates creates a new mock instance. +func NewMockGates(ctrl *gomock.Controller) *MockGates { + mock := &MockGates{ctrl: ctrl} + mock.recorder = &MockGatesMockRecorder{mock} return mock } // EXPECT returns an object that allows the caller to indicate expected use. -func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { +func (m *MockGates) EXPECT() *MockGatesMockRecorder { return m.recorder } // For mocks base method. -func (m *MockFactory) For(cfg admissiongate.Config) (admissiongate.Gate, error) { +func (m *MockGates) For(point admissiongate.Point, request entity.Request) (admissiongate.Gate, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "For", cfg) + ret := m.ctrl.Call(m, "For", point, request) ret0, _ := ret[0].(admissiongate.Gate) ret1, _ := ret[1].(error) return ret0, ret1 } // For indicates an expected call of For. -func (mr *MockFactoryMockRecorder) For(cfg any) *gomock.Call { +func (mr *MockGatesMockRecorder) For(point, request any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), cfg) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates)(nil).For), point, request) } From 51a24c41ea717d4315a33ccf8463503d56c5ce08 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Thu, 17 Sep 2026 20:26:52 +0000 Subject: [PATCH 5/5] refactor(stovepipe): bind admission purpose in wiring --- doc/rfc/stovepipe/admission-gate.md | 26 +++++++------------ stovepipe/extension/admissiongate/README.md | 4 +-- .../extension/admissiongate/admissiongate.go | 24 +++++------------ .../admissiongate/mock/admissiongate_mock.go | 8 +++--- 4 files changed, 23 insertions(+), 39 deletions(-) diff --git a/doc/rfc/stovepipe/admission-gate.md b/doc/rfc/stovepipe/admission-gate.md index ade871f77..14cd30dbc 100644 --- a/doc/rfc/stovepipe/admission-gate.md +++ b/doc/rfc/stovepipe/admission-gate.md @@ -13,11 +13,11 @@ The immediate requirements are: - After a build runner reports a failed result, defer the next admission for a configured cooldown. - Keep coalescing active while a request is deferred, so a newer head can supersede it without waiting for the gate to open. -The framework must also leave room for policies such as maintenance windows, resource budgets, provider health, or an operator hold, and for logical admission points other than build admission. +The framework must also leave room for policies such as maintenance windows, resource budgets, provider health, or an operator hold, and for logical admission boundaries other than build admission. ## Scope -An **admission gate** decides whether one domain entity may cross a logical pipeline boundary. The first point is `build`: whether the latest accepted Stovepipe request may reserve validation capacity and advance toward the build stage. +An **admission gate** decides whether one domain entity may cross a logical pipeline boundary. The first use is build admission: whether the latest accepted Stovepipe request may reserve validation capacity and advance toward the build stage. This is separate from the shared [Consumer Gate](../consumer-gate.md). A consumer gate is an external operational control that stops deliveries before a controller. A Stovepipe admission gate is domain policy evaluated by a controller for a specific request. Both defer with queue redelivery, but they answer different questions and own different state. @@ -26,10 +26,6 @@ This is separate from the shared [Consumer Gate](../consumer-gate.md). A consume The vendor-neutral contract lives at `stovepipe/extension/admissiongate`: ```go -type Point string - -const PointBuild Point = "build" - type Result struct { Decision Decision BlockedBy []string @@ -40,13 +36,11 @@ type Gate interface { } type Gates interface { - For(Point, entity.Request) (Gate, error) + For(entity.Request) (Gate, error) } ``` -`Point` is an open string identifier. The shared package names only points understood by Stovepipe; adding a point is additive and does not change `Gate`. - -`Gates` is the host-owned resolver across queues and points. `For` takes the request rather than a separate queue configuration because the request already carries its authoritative queue identity. It returns exactly one composite gate: returning a slice of independently stateful gates would make atomic admission impossible when one gate records a reservation before a later gate defers. Concrete routing belongs in service wiring, not an extension implementation package. +`Gates` is the host-owned resolver across queues for one logical boundary. The controller receives the resolver for the admission it performs, so that context does not need to travel through a string identifier on every call. `For` takes the request rather than a separate queue configuration because the request already carries its authoritative queue identity. It returns exactly one composite gate: returning a slice of independently stateful gates would make atomic admission impossible when one gate records a reservation before a later gate defers. Concrete routing belongs in service wiring, not an extension implementation package. `TryAdmit` takes the thin `entity.Request`, following the repository's identity-in extension rule. The request already identifies its queue. A gate resolves the queue-scoped storage, configuration, request history, clocks, or remote services it needs through dependencies injected when its implementation is constructed. Controllers do not pre-resolve policy facts and hand them across the contract. @@ -65,7 +59,7 @@ There is intentionally no `Complete`, `Release`, or `RecordOutcome` method. `Try `process` retains responsibility for request choreography; the gate owns only admission policy and its reservation: 1. Load the request and queue, then coalesce it against the latest request ID. -2. Resolve the gate with `gates.For(PointBuild, request)`. +2. Resolve the gate with `gates.For(request)`. 3. Call `TryAdmit(ctx, request)`. 4. On `DecisionDeferred`, hold the delivery for the queue's normal gate re-check delay and return successfully. 5. On redelivery, start again at coalescing before evaluating the gate. @@ -127,7 +121,7 @@ Using request history is what lets failure cooldown mean "after the runner-repor ## Policy Composition -One resolved `Gate` is the atomic composition boundary for one queue and admission point. The implementation may contain several policies, but they are not independently stateful extensions called in sequence. +One resolved `Gate` is the atomic composition boundary for one queue and logical admission. The implementation may contain several policies, but they are not independently stateful extensions called in sequence. For each `TryAdmit`, the implementation: @@ -145,19 +139,19 @@ The initial standard gate composes: - **Minimum interval:** when configured above zero, require at least that many milliseconds between admission timestamps. Non-positive values disable it. - **Failure cooldown:** after a configured runner-reported failure, defer until the failure occurrence time plus the cooldown. Non-positive values disable it. -Future policies at the same point, such as a calendar window, cost budget, provider-health circuit, or manual hold, fit inside the same atomic composition. A policy that needs its own durable facts receives a namespaced section in the gate envelope. A fundamentally different backend or evaluation model is another `Gate` implementation selected by wiring. +Future policies for build admission, such as a calendar window, cost budget, provider-health circuit, or manual hold, fit inside the same atomic composition. A policy that needs its own durable facts receives a namespaced section in the gate envelope. A fundamentally different backend or evaluation model is another `Gate` implementation selected by wiring. ## Configuration And Routing Queue policy settings remain deployment configuration supplied through `queueconfig`; mutable observations remain in `AdmissionState`. Configuration is read during evaluation so a changed interval or cooldown affects the next attempt without rewriting stored state. -The common admission-gate contract does not define a universal policy configuration language. The standard gate understands Stovepipe's typed queue settings. Another gate may receive configuration through dependencies injected by its constructor. Per-queue and per-point selection belongs in service wiring through `Gates.For`, consistent with other plural resolver contracts; no implementation package contains a routing map. +The common admission-gate contract does not define a universal policy configuration language. The standard gate understands Stovepipe's typed queue settings. Another gate may receive configuration through dependencies injected by its constructor. Per-queue selection belongs in service wiring through `Gates.For`, consistent with other plural resolver contracts; no implementation package contains a routing map. -This separation permits gradual evolution. The current build point can use the standard composite gate, a high-cost queue can use a remote budget gate, and a future promotion point can use an independently selected gate without changing controllers' result handling. +This separation permits gradual evolution. Build admission can use the standard composite gate, a high-cost queue can use a remote budget gate, and a future promotion controller can receive a separately wired `Gates` resolver without changing the gate or result contracts. ## Observability -The controller records admitted and deferred counters tagged by admission point. Deferred counters may additionally use each `BlockedBy` identifier, whose low-cardinality contract makes it safe as a metric tag. Logs include request ID, queue, point, decision, and blockers. +The controller records admitted and deferred counters; its operation name identifies the guarded boundary. Deferred counters may additionally use each `BlockedBy` identifier, whose low-cardinality contract makes it safe as a metric tag. Logs include request ID, queue, decision, and blockers. The gate implementation records evaluation, state decode, reconciliation, CAS-conflict, and dependency errors. It must not place opaque state contents or arbitrary configuration values in metric tags. diff --git a/stovepipe/extension/admissiongate/README.md b/stovepipe/extension/admissiongate/README.md index c334f540a..bea1f256c 100644 --- a/stovepipe/extension/admissiongate/README.md +++ b/stovepipe/extension/admissiongate/README.md @@ -1,5 +1,5 @@ # Admission Gate Extension -Vendor-neutral contract for deciding whether a Stovepipe request may cross a logical admission point. See the [Stovepipe Admission Gates RFC](../../../doc/rfc/stovepipe/admission-gate.md) for decision semantics, durable-state ownership, composition, and controller integration. +Vendor-neutral contract for deciding whether a Stovepipe request may cross a logical pipeline boundary. See the [Stovepipe Admission Gates RFC](../../../doc/rfc/stovepipe/admission-gate.md) for decision semantics, durable-state ownership, composition, and controller integration. -Implementations take request identity, resolve their own facts, and return an admitted or deferred result. The `Gates` resolver selects one composite gate by request and admission point; its concrete routing belongs in service wiring. +Implementations take request identity, resolve their own facts, and return an admitted or deferred result. A controller receives a `Gates` resolver for its boundary, and that resolver selects one composite gate by request; concrete queue routing belongs in service wiring. diff --git a/stovepipe/extension/admissiongate/admissiongate.go b/stovepipe/extension/admissiongate/admissiongate.go index afcf80b4f..1f3241288 100644 --- a/stovepipe/extension/admissiongate/admissiongate.go +++ b/stovepipe/extension/admissiongate/admissiongate.go @@ -13,7 +13,7 @@ // limitations under the License. // Package admissiongate defines Stovepipe's extension contract for deciding -// whether a request may cross a logical pipeline admission point. +// whether a request may cross a logical pipeline boundary. package admissiongate //go:generate mockgen -source=admissiongate.go -destination=mock/admissiongate_mock.go -package=mock @@ -24,16 +24,6 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" ) -// Point identifies a logical pipeline boundary guarded by an admission gate. -type Point string - -const ( - // PointUnknown is the invalid zero value. - PointUnknown Point = "" - // PointBuild guards admission of a request toward the build stage. - PointBuild Point = "build" -) - // Decision is the expected outcome of evaluating an admission gate. type Decision string @@ -54,7 +44,7 @@ type Result struct { BlockedBy []string } -// Gate decides whether requests may cross one queue-scoped admission point. +// Gate decides whether requests may cross one queue-scoped pipeline boundary. // Implementations resolve the durable facts they need from the request's // identity and must make repeated calls for the same request idempotent. type Gate interface { @@ -64,10 +54,10 @@ type Gate interface { TryAdmit(ctx context.Context, request entity.Request) (Result, error) } -// Gates resolves the gate for a request and logical admission point. Concrete -// routing belongs in service wiring rather than an extension implementation -// package. +// Gates resolves the gate for a request. A controller receives the resolver +// for the pipeline boundary it owns; concrete queue routing belongs in service +// wiring rather than an extension implementation package. type Gates interface { - // For returns the Gate selected for point and request. - For(point Point, request entity.Request) (Gate, error) + // For returns the Gate selected for request. + For(request entity.Request) (Gate, error) } diff --git a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go b/stovepipe/extension/admissiongate/mock/admissiongate_mock.go index ede85e46a..a2a4bdb44 100644 --- a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go +++ b/stovepipe/extension/admissiongate/mock/admissiongate_mock.go @@ -82,16 +82,16 @@ func (m *MockGates) EXPECT() *MockGatesMockRecorder { } // For mocks base method. -func (m *MockGates) For(point admissiongate.Point, request entity.Request) (admissiongate.Gate, error) { +func (m *MockGates) For(request entity.Request) (admissiongate.Gate, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "For", point, request) + ret := m.ctrl.Call(m, "For", request) ret0, _ := ret[0].(admissiongate.Gate) ret1, _ := ret[1].(error) return ret0, ret1 } // For indicates an expected call of For. -func (mr *MockGatesMockRecorder) For(point, request any) *gomock.Call { +func (mr *MockGatesMockRecorder) For(request any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates)(nil).For), point, request) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates)(nil).For), request) }