From 5e12652d442f0f16f75127ff3966636f4792f0da Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 21 Sep 2026 15:21:48 +0000 Subject: [PATCH 1/4] docs(stovepipe): define admission gate framework Summary: Intent: - Define an extensible model for queue-scoped logical admission before implementation. - Keep policy composition atomic while supporting future admission boundaries. Changes: - Document gate semantics, opaque state ownership, reconciliation, composition, and rollout. - Add the request-based admissiongate contract, Gates resolver, generated mocks, and Bazel targets. Test Plan: - make lint - make check-gazelle - make check-tidy - ./tool/bazel build //stovepipe/extension/admissiongate/... Revert Plan: Revert this PR to remove the proposed admission-gate contract and RFC. --- Makefile | 2 +- doc/rfc/index.md | 1 + doc/rfc/stovepipe/admission-gate.md | 186 ++++++++++++++++++ stovepipe/extension/admissiongate/BUILD.bazel | 9 + stovepipe/extension/admissiongate/README.md | 5 + .../extension/admissiongate/admissiongate.go | 63 ++++++ .../extension/admissiongate/mock/BUILD.bazel | 13 ++ .../admissiongate/mock/admissiongate_mock.go | 97 +++++++++ 8 files changed, 375 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..14cd30dbc --- /dev/null +++ b/doc/rfc/stovepipe/admission-gate.md @@ -0,0 +1,186 @@ +# 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 boundaries other than build admission. + +## Scope + +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. + +## Contract + +The vendor-neutral contract lives at `stovepipe/extension/admissiongate`: + +```go +type Result struct { + Decision Decision + BlockedBy []string +} + +type Gate interface { + TryAdmit(context.Context, entity.Request) (Result, error) +} + +type Gates interface { + For(entity.Request) (Gate, error) +} +``` + +`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. + +`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 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. +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 logical admission. 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 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 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. 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; 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. + +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 `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. + +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..bea1f256c --- /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 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. 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 new file mode 100644 index 000000000..1f3241288 --- /dev/null +++ b/stovepipe/extension/admissiongate/admissiongate.go @@ -0,0 +1,63 @@ +// 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 boundary. +package admissiongate + +//go:generate mockgen -source=admissiongate.go -destination=mock/admissiongate_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// 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 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 { + // 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) +} + +// 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 request. + For(request entity.Request) (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..a2a4bdb44 --- /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) +} + +// MockGates is a mock of Gates interface. +type MockGates struct { + ctrl *gomock.Controller + recorder *MockGatesMockRecorder + isgomock struct{} +} + +// MockGatesMockRecorder is the mock recorder for MockGates. +type MockGatesMockRecorder struct { + mock *MockGates +} + +// 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 *MockGates) EXPECT() *MockGatesMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockGates) For(request entity.Request) (admissiongate.Gate, error) { + m.ctrl.T.Helper() + 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(request any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates)(nil).For), request) +} From 8a25a54080ec248f015e61d86ee6ebe7993ee946 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 21 Sep 2026 15:38:38 +0000 Subject: [PATCH 2/4] docs(stovepipe): align admission RFC with main --- doc/rfc/stovepipe/admission-gate.md | 28 +++++++++++++++++----------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/doc/rfc/stovepipe/admission-gate.md b/doc/rfc/stovepipe/admission-gate.md index 14cd30dbc..8489d2788 100644 --- a/doc/rfc/stovepipe/admission-gate.md +++ b/doc/rfc/stovepipe/admission-gate.md @@ -4,7 +4,7 @@ Status: proposed. This RFC defines the framework and extension contract; storage ## 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. +Stovepipe currently implements one build-admission policy directly in `process`: admit only while the queue's in-flight count is below its concurrency limit. General admission throttling and a cooldown after a failed build are new requirements. Implementing each new rule in a controller would spread policy across lifecycle stages and make every additional rule another special case. The immediate requirements are: @@ -15,6 +15,12 @@ The immediate requirements are: 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. +## Current Behavior + +On `main`, `process` loads the queue and its `QueueConfig`, coalesces the request against `Queue.LatestRequestID`, and compares `Queue.InFlightCount` with `QueueConfig.MaxConcurrent`. When the queue is full, it calls `delivery.Hold(QueueConfig.GateWaitDelayMs)` and returns successfully. `GateWaitDelayMs` is only the delay before redelivery and another concurrency check; it does not impose a minimum interval between admitted builds. Redelivery starts from request loading and coalescing, so a newer head can supersede the deferred request. + +When capacity is available, `process` claims it by incrementing `Queue.InFlightCount` with a queue-version CAS. A queue version conflict reloads the queue and repeats coalescing before another claim. After a terminal runner result, `buildsignal` decrements the counter before marking the request terminal. Relevant DLQ paths also decrement it so abandoned processing work does not permanently consume capacity. There is no minimum-admission timestamp, failure-cooldown state, or opaque admission payload on the queue today. + ## Scope 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. @@ -56,7 +62,7 @@ There is intentionally no `Complete`, `Release`, or `RecordOutcome` method. `Try ## Process Integration -`process` retains responsibility for request choreography; the gate owns only admission policy and its reservation: +`process` retains responsibility for request choreography; the gate owns only admission policy and its reservation. The integration preserves the current re-check behavior: 1. Load the request and queue, then coalesce it against the latest request ID. 2. Resolve the gate with `gates.For(request)`. @@ -117,7 +123,7 @@ The lookup cost is bounded by the configured concurrency limit rather than queue 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. +This reconciliation replaces the current shared-counter ownership in which `process` increments `Queue.InFlightCount` and `buildsignal` or a DLQ path decrements it. Under the proposal, `buildsignal` remains policy-neutral: it persists the build result, request terminal state, and request log, but neither understands the gate envelope nor invokes an admission callback. The next `TryAdmit` observes those durable facts and removes completed reservations. ## Policy Composition @@ -135,15 +141,15 @@ This prevents partial admission: an interval policy cannot consume its next slot 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. +- **Concurrency:** preserve the current rule by deferring while the number of reconciled active admissions is at the per-queue limit. +- **Minimum interval:** add general throttling by requiring at least the configured number of milliseconds between admission timestamps. Non-positive values disable it. +- **Failure cooldown:** add outcome-sensitive throttling by deferring until a runner-reported failure's occurrence time plus the configured cooldown. Non-positive values disable it. 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. +Today `queueconfig` supplies `MaxConcurrent` and `GateWaitDelayMs`, and the service uses its built-in default implementation. The proposed minimum-interval and failure-cooldown settings belong in the same typed queue configuration contract; a deployment-backed configuration implementation is separate implementation work. Mutable observations such as admission times and failure facts belong 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 selection belongs in service wiring through `Gates.For`, consistent with other plural resolver contracts; no implementation package contains a routing map. @@ -168,10 +174,10 @@ Unknown state versions also fail closed. Silently resetting an unreadable payloa 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 `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. +2. Implement the standard composite gate with a concurrency policy matching the current `InFlightCount < MaxConcurrent` behavior. +3. Wire `process` through `Gates`, replace its direct counter claim, and make reconciliation authoritative instead of direct releases from `buildsignal` and DLQ paths. +4. Add minimum interval as a new standard policy for general admission throttling. +5. Add failure cooldown as another new 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. From 57e8ed645776bf8509907f76f036e944409ac8b7 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 21 Sep 2026 16:01:03 +0000 Subject: [PATCH 3/4] docs(stovepipe): refine admission gate contract --- doc/rfc/stovepipe/admission-gate.md | 60 ++++++++++--------- stovepipe/extension/admissiongate/README.md | 2 +- .../extension/admissiongate/admissiongate.go | 17 ++++-- .../admissiongate/mock/admissiongate_mock.go | 8 +-- 4 files changed, 51 insertions(+), 36 deletions(-) diff --git a/doc/rfc/stovepipe/admission-gate.md b/doc/rfc/stovepipe/admission-gate.md index 8489d2788..3ac657947 100644 --- a/doc/rfc/stovepipe/admission-gate.md +++ b/doc/rfc/stovepipe/admission-gate.md @@ -4,7 +4,7 @@ Status: proposed. This RFC defines the framework and extension contract; storage ## Problem -Stovepipe currently implements one build-admission policy directly in `process`: admit only while the queue's in-flight count is below its concurrency limit. General admission throttling and a cooldown after a failed build are new requirements. Implementing each new rule in a controller would spread policy across lifecycle stages and make every additional rule another special case. +Stovepipe currently implements one build-admission policy directly in `process`: admit only while the queue's in-flight count is below its concurrency limit. New requirements, such as general admission throttling and a cooldown after a failed build, will continue to surface, indicating a need for a standard admission gating framework that individual controllers can use to gate work based on customized policy decisions. The immediate requirements are: @@ -15,12 +15,6 @@ The immediate requirements are: 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. -## Current Behavior - -On `main`, `process` loads the queue and its `QueueConfig`, coalesces the request against `Queue.LatestRequestID`, and compares `Queue.InFlightCount` with `QueueConfig.MaxConcurrent`. When the queue is full, it calls `delivery.Hold(QueueConfig.GateWaitDelayMs)` and returns successfully. `GateWaitDelayMs` is only the delay before redelivery and another concurrency check; it does not impose a minimum interval between admitted builds. Redelivery starts from request loading and coalescing, so a newer head can supersede the deferred request. - -When capacity is available, `process` claims it by incrementing `Queue.InFlightCount` with a queue-version CAS. A queue version conflict reloads the queue and repeats coalescing before another claim. After a terminal runner result, `buildsignal` decrements the counter before marking the request terminal. Relevant DLQ paths also decrement it so abandoned processing work does not permanently consume capacity. There is no minimum-admission timestamp, failure-cooldown state, or opaque admission payload on the queue today. - ## Scope 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. @@ -32,40 +26,52 @@ This is separate from the shared [Consumer Gate](../consumer-gate.md). A consume The vendor-neutral contract lives at `stovepipe/extension/admissiongate`: ```go +type Blocker string + type Result struct { Decision Decision - BlockedBy []string + BlockedBy []Blocker } type Gate interface { TryAdmit(context.Context, entity.Request) (Result, error) } +type Config struct { + QueueName string +} + type Gates interface { - For(entity.Request) (Gate, error) + For(Config) (Gate, error) } ``` -`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. +`Gates` is the host-owned resolver across queues for one logical boundary. The controller receives the resolver for the admission it performs. Like the `buildrunner`, `sourcecontrol`, and `storage` resolver contracts, `For` takes a typed `Config` containing only `QueueName`; wiring uses that identity to select and bind the implementation. 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. +`TryAdmit` takes the thin `entity.Request`, following the repository's identity-in extension rule for request-stage decisions. Passing only its string ID would discard the queue and immutable request identity already available to the controller, force implementations to parse an ID or add another lookup merely to recover that identity, and diverge from other decision extensions that receive the stage entity. The request is a reference, not a bundle of pre-resolved policy facts: a gate reloads mutable state and resolves queue-scoped storage, configuration, request history, clocks, or remote services through dependencies injected when its implementation is constructed. `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. +- `DecisionDeferred` means no admission was recorded because one or more policies currently block it. `BlockedBy` contains typed `Blocker` values: stable, low-cardinality policy identifiers for logs and metrics, not human-readable errors. Each implementation defines constants for the policies it can report. +- `DecisionUnknown` is not a runtime outcome. It is the invalid zero value, consistent with entity enums elsewhere in the repository, so an implementation that accidentally returns an empty `Result` fails closed. The controller converts a nil-error result with this decision into an error. 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. +## Current Behavior + +On `main`, `process` loads the queue and its `QueueConfig`, coalesces the request against `Queue.LatestRequestID`, and compares `Queue.InFlightCount` with `QueueConfig.MaxConcurrent`. When the queue is full, it calls `delivery.Hold(QueueConfig.GateWaitDelayMs)` and returns successfully. `GateWaitDelayMs` is only the delay before redelivery and another concurrency check; it does not impose a minimum interval between admitted builds. Redelivery starts from request loading and coalescing, so a newer head can supersede the deferred request. + +When capacity is available, `process` claims it by incrementing `Queue.InFlightCount` with a queue-version CAS. A queue version conflict reloads the queue and repeats coalescing before another claim. After a terminal runner result, `buildsignal` decrements the counter before marking the request terminal. Relevant DLQ paths also decrement it so abandoned processing work does not permanently consume capacity. There is no minimum-admission timestamp, failure-cooldown state, or opaque admission payload on the queue today. + ## Process Integration `process` retains responsibility for request choreography; the gate owns only admission policy and its reservation. The integration preserves the current re-check behavior: 1. Load the request and queue, then coalesce it against the latest request ID. -2. Resolve the gate with `gates.For(request)`. +2. Resolve the gate with `gates.For(admissiongate.Config{QueueName: request.Queue})`. 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. @@ -75,36 +81,36 @@ The hold delay belongs to controller scheduling configuration, not `Result`. A p 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 +## State And Storage + +The extension contract does not prescribe storage. A stateless gate stores nothing; another implementation may use an implementation-owned table, a key-value store, or a remote quota service. Those dependencies are injected when the implementation is constructed, and neither `Gates` nor `Gate` exposes a generic state API. -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 proposed standard composite build gate does need a small amount of durable state for idempotent reservations, concurrency reconciliation, minimum-interval history, and failure cooldown. Its 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. +Keeping this implementation's state on the queue is deliberate rather than a framework requirement. The standard build gate must order its reservation with `Queue.LatestRequestID`, matching the current queue CAS that prevents a newly superseded head from claiming capacity. A separate table would give admission state its own CAS but could not atomically observe the latest-head update without a cross-entity transaction. A different gate whose facts do not need that ordering should own its own table or backend instead of adding data to this payload. + +The standard gate exclusively owns the payload's encoding and compatibility. Its 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. Another gate does not read or write this envelope. Changing the standard 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 - } - }, + "active_request_ids": [ + "request/monorepo/main/42" + ], "policies": { "minimum_interval": { "last_admitted_at_ms": 1789506000000 }, "failure_cooldown": { - "not_before_ms": 1789509600000, - "source_request_id": "request/monorepo/main/41" + "not_before_ms": 1789509600000 } } } ``` -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. +This shape illustrates the minimum facts the initial policies need, not a shared wire contract. Active request IDs support idempotency and concurrency reconciliation; the last admission time survives completion for general throttling; and the cooldown deadline survives removal of the failed request. 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. @@ -112,7 +118,7 @@ No cross-entity transaction is introduced. Recording an admission and transition ## 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: +The state retains the IDs 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. diff --git a/stovepipe/extension/admissiongate/README.md b/stovepipe/extension/admissiongate/README.md index bea1f256c..6034a2eaa 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 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. 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. +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 from a queue-scoped `Config`; concrete queue routing belongs in service wiring. diff --git a/stovepipe/extension/admissiongate/admissiongate.go b/stovepipe/extension/admissiongate/admissiongate.go index 1f3241288..eebc72589 100644 --- a/stovepipe/extension/admissiongate/admissiongate.go +++ b/stovepipe/extension/admissiongate/admissiongate.go @@ -36,12 +36,15 @@ const ( DecisionDeferred Decision = "deferred" ) +// Blocker identifies a policy that currently prevents admission. +type Blocker string + // 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 + BlockedBy []Blocker } // Gate decides whether requests may cross one queue-scoped pipeline boundary. @@ -54,10 +57,16 @@ type Gate interface { TryAdmit(ctx context.Context, request entity.Request) (Result, error) } -// Gates resolves the gate for a request. A controller receives the resolver +// Config identifies the queue for which a Gate is resolved. +type Config struct { + // QueueName identifies the queue the resolved Gate serves. + QueueName string +} + +// Gates resolves the gate for a queue. 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 request. - For(request entity.Request) (Gate, error) + // For returns the Gate selected for config. + For(config Config) (Gate, error) } diff --git a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go b/stovepipe/extension/admissiongate/mock/admissiongate_mock.go index a2a4bdb44..133ba8ef3 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(request entity.Request) (admissiongate.Gate, error) { +func (m *MockGates) For(config admissiongate.Config) (admissiongate.Gate, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "For", request) + ret := m.ctrl.Call(m, "For", config) ret0, _ := ret[0].(admissiongate.Gate) ret1, _ := ret[1].(error) return ret0, ret1 } // For indicates an expected call of For. -func (mr *MockGatesMockRecorder) For(request any) *gomock.Call { +func (mr *MockGatesMockRecorder) For(config any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates)(nil).For), request) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates)(nil).For), config) } From 8170db01b1b1add0631ef8a399bce283808609cd Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 22 Sep 2026 20:09:52 +0000 Subject: [PATCH 4/4] docs(platform): generalize admission gate contract --- Makefile | 2 +- doc/rfc/admission-gate.md | 145 +++++++++++++ doc/rfc/index.md | 2 +- doc/rfc/stovepipe/admission-gate.md | 198 ------------------ .../extension/admissiongate/BUILD.bazel | 3 +- platform/extension/admissiongate/README.md | 5 + .../extension/admissiongate/admissiongate.go | 40 ++-- .../extension/admissiongate/mock/BUILD.bazel | 5 +- .../admissiongate/mock/admissiongate_mock.go | 51 +++-- stovepipe/extension/admissiongate/README.md | 5 - 10 files changed, 199 insertions(+), 257 deletions(-) create mode 100644 doc/rfc/admission-gate.md delete mode 100644 doc/rfc/stovepipe/admission-gate.md rename {stovepipe => platform}/extension/admissiongate/BUILD.bazel (55%) create mode 100644 platform/extension/admissiongate/README.md rename {stovepipe => platform}/extension/admissiongate/admissiongate.go (59%) rename {stovepipe => platform}/extension/admissiongate/mock/BUILD.bazel (55%) rename {stovepipe => platform}/extension/admissiongate/mock/admissiongate_mock.go (54%) delete mode 100644 stovepipe/extension/admissiongate/README.md diff --git a/Makefile b/Makefile index 448da31b8..f36be69a5 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/admissiongate/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... + @$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/admissiongate/... ./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/... @echo "Mocks generated successfully!" proto: ## Generate protobuf files from .proto definitions diff --git a/doc/rfc/admission-gate.md b/doc/rfc/admission-gate.md new file mode 100644 index 000000000..4c1936c38 --- /dev/null +++ b/doc/rfc/admission-gate.md @@ -0,0 +1,145 @@ +# Controller Admission Gates + +## Scope + +An **admission gate** decides whether a domain entity may cross a queue-scoped logical pipeline boundary. The shared contract is typed to that entity; each use supplies its own policies, dependencies, and any storage it needs. Gates may enforce concurrency, rate limits, cooldowns, maintenance windows, resource budgets, provider health, operator holds, or other admission policies. + +This differs from the shared [Consumer Gate](consumer-gate.md). A consumer gate is an external operational control checked before a controller receives a delivery. An admission gate is domain policy evaluated inside a controller. Both may defer work through redelivery, but they answer different questions and own different state. + +## Contract + +The vendor-neutral contract lives at `platform/extension/admissiongate`: + +```go +type Blocker string + +type Result struct { + Decision Decision + BlockedBy []Blocker +} + +type Gate[T any] interface { + TryAdmit(context.Context, T) (Result, error) +} + +type Config struct { + QueueName string +} + +type Gates[T any] interface { + For(Config) (Gate[T], error) +} +``` + +`Gate[T]` preserves domain identity without making `platform` depend on a domain. Each use binds `T` to its native stage entity, such as a request or batch. The platform package contains only this contract and its generated mocks. Implementations remain with their owning domains. + +`Gates[T]` represents one logical boundary and resolves its composite gate for a queue. `For` follows existing resolver patterns: `Config` carries only `QueueName`, and service wiring selects and binds the implementation. It returns one gate rather than exposing the implementation's policy composition to the controller. + +`TryAdmit` receives the owning domain's thin stage entity, following the repository's identity-in rule for decision extensions. Passing only an ID would discard identity already available to the controller and force the implementation to parse or reload it. The entity remains a reference: the implementation reloads mutable facts and resolves other dependencies internally. + +`Result` expresses expected control flow: + +- `DecisionAdmitted` means current policy permits the controller to continue admitting the candidate. The controller may still need to claim domain capacity before advancing it. +- `DecisionDeferred` means policy currently blocks the candidate. `BlockedBy` contains typed, low-cardinality policy identifiers for logs and metrics. Implementations define constants for the policies they report. +- `DecisionUnknown` follows the repository's enum convention as the invalid zero value, not a runtime outcome. The controller converts a nil-error result with this decision into an error, so an empty `Result` fails closed. + +Errors are reserved for policy-evaluation failures. Deferral is not an error and does not consume retry budget. + +There is no `Complete`, `Release`, or `RecordOutcome` method. `TryAdmit` evaluates authoritative facts through dependencies supplied to the implementation. Domain capacity and lifecycle transitions remain owned by their existing controllers. + +## State And Storage + +The shared contract does not prescribe or require storage. A gate may evaluate configuration and existing domain facts directly or delegate to a policy service. Dependencies are injected at construction; `Gate[T]` exposes no generic state API. + +Policy state belongs to the dependency that makes the policy authoritative, not to the admission framework. A static policy may need no state. A policy that consumes a rate or budget must provide an atomic, candidate-idempotent decision so concurrent evaluations cannot spend the same capacity twice. Its concrete backend owns any schema or remote storage it needs. + +The framework adds neither policy-specific columns nor an opaque metadata field to domain tables. A concrete policy may reuse existing domain state when that state already expresses the required fact, but the gate does not take ownership of that state. + +## Reconciliation + +Admission does not replace domain reconciliation. Controllers continue to own capacity claims, releases, and lifecycle facts. A gate reads those facts or delegates to a policy provider that maintains its own authoritative view. + +Outcome-sensitive policies should derive decisions from durable request state and history when practical. The initial Stovepipe cooldown reacts only to `RequestOutcomeReasonBuildFailed`; success, cancellation, superseding, and DLQ- or timeout-synthesized failures do not activate it. If efficient evaluation requires derived state, the concrete policy owns that projection and its reconciliation. + +## Policy Composition + +One `Gate[T]` represents the composed policy decision for a queue and logical admission. Policy primitives remain implementation details, so an implementation may evaluate local rules, call a remote policy service, or use a purpose-built algorithm. + +Pure policies may be evaluated together without coordination. Policies that consume capacity must expose an atomic composite operation or define compensation; the shared gate contract does not invent a transaction across independent policy backends. Any capacity-consuming operation must be idempotent for the candidate because queue delivery is at least once. + +The initial Stovepipe admission flow combines: + +- **Concurrency:** `process` retains the existing queue-counter check and CAS claim. +- **Minimum interval:** the gate requires the configured number of milliseconds between admissions; non-positive disables it. +- **Failure cooldown:** the gate defers until a runner-reported failure's occurrence time plus the configured cooldown; non-positive disables it. + +Future Stovepipe policies remain behind the same gate. A fundamentally different backend or evaluation model is another `Gate[entity.Request]` selected by wiring. + +## Configuration And Routing + +Today `queueconfig` supplies `MaxConcurrent` and `GateWaitDelayMs` through its built-in default implementation. The proposed minimum interval and failure cooldown belong in the same typed queue configuration. Configuration changes affect the next evaluation; mutable observations remain in the domain or policy provider that owns them. + +The shared contract defines no universal policy configuration language. Implementations receive configuration and other dependencies at construction. Per-queue routing belongs in service wiring through `Gates[T].For`, not in implementation packages. + +## Observability + +The controller records admitted and deferred counters; its operation name identifies the guarded boundary. Deferred counters may use `BlockedBy` as a low-cardinality dimension. Logs include candidate ID, queue, decision, and blockers. + +The implementation records evaluation and dependency errors. Arbitrary configuration values and policy state must not appear in metric tags. + +## Failure Posture + +Admission fails closed. If configuration or policy evaluation fails, `TryAdmit` returns an error and the controller does not advance the candidate. Normal consumer retry and DLQ behavior handles persistent failures. + +## Rejected Alternatives + +- **Callbacks from terminal paths:** `Complete` or `RecordOutcome` would make domain lifecycle controllers depend on the gate. Policies that need outcomes consume authoritative domain facts or maintain their own projection. +- **One extension call per policy:** exposing policy composition to the controller couples choreography to policy configuration. One resolved gate returns the composed decision. +- **Gate-owned domain columns:** policy-specific fields or an opaque gate payload would couple replaceable policy implementations to shared domain storage. +- **Sleeping until a policy deadline:** long holds suppress coalescing. Short re-checks preserve responsive superseding. +- **Process-local policy state:** replicas diverge and restarts lose rate, budget, or cooldown facts. Stateful policies require an authoritative backend. +- **Failing open:** policy infrastructure failure must not silently remove protection for finite or costly resources. + +## Example Use Case: Stovepipe Build Gating + +### Use Case + +Stovepipe build admission is the first use of the shared contract. Its `process` stage must decide whether the latest accepted request may reserve validation capacity and advance to `build`. + +Stovepipe currently enforces only the queue's concurrency limit, directly in `process`. General throttling, failure cooldowns, and future policies need a common framework instead of more controller-specific branches. The immediate requirements are: + +- Limit concurrent logical validations per queue. +- Throttle admissions per queue, for example to one start per hour. +- After the build runner reports a failure, defer the next admission for a configured cooldown. +- Continue coalescing while a request is deferred so a newer head can supersede it promptly. + +On `main`, `process` loads the queue and `QueueConfig`, coalesces the request against `Queue.LatestRequestID`, and compares `Queue.InFlightCount` with `QueueConfig.MaxConcurrent`. A full queue causes `delivery.Hold(QueueConfig.GateWaitDelayMs)` followed by a successful return. `GateWaitDelayMs` is only the redelivery cadence; it is not a minimum interval between builds. Redelivery starts with loading and coalescing again, so a newer head can supersede the deferred request. + +When capacity is available, `process` increments `Queue.InFlightCount` with a queue-version CAS. A version conflict reloads the queue and repeats coalescing before another claim. After a terminal runner result, `buildsignal` decrements the counter before making the request terminal. Relevant DLQ paths also decrement it. There is no admission timestamp, failure-cooldown state, or opaque admission payload today. + +### Integration + +`process` retains request choreography and build-slot ownership; the gate owns only the policy decision: + +1. Load the request and queue, then coalesce against the latest request ID. +2. Resolve `Gate[entity.Request]` with `gates.For(admissiongate.Config{QueueName: request.Queue})`. +3. Call `TryAdmit(ctx, request)`. +4. On `DecisionDeferred`, hold for the normal gate re-check delay and return successfully. +5. On `DecisionAdmitted`, run the existing concurrency check, derive the build strategy, and CAS-increment `Queue.InFlightCount` to claim a build slot. +6. If the queue CAS conflicts, reload, coalesce, and reevaluate policy and capacity before retrying. +7. After claiming a slot, transition the request to `processing` and publish to `build` using the existing persist-before-publish ordering. + +On redelivery, `process` restarts at loading and coalescing. The hold delay remains controller scheduling configuration rather than part of `Result`. Waiting until a policy's exact deadline would suppress coalescing for that entire period; short bounded re-checks keep superseding responsive. + +The existing slot lifecycle remains unchanged. `process` compensates if its slot claim is not followed by the transition to `processing`; `buildsignal` releases the slot before recording a terminal build outcome; and request, build, and buildsignal DLQ reconciliation releases a slot when failing a `processing` request. The gate receives no lifecycle callbacks. + +### Rollout + +The Stovepipe integration can migrate incrementally: + +1. Land the shared contract and a Stovepipe gate implementation. +2. Wire `process` through `Gates[entity.Request]` while preserving its existing concurrency claim and release paths. +3. Add minimum interval as a general-throttling policy. +4. Add failure cooldown as an outcome-sensitive policy. + +The framework requires no queue-schema migration. A stateful policy introduces and owns its backend only when that policy is implemented; the admission-gate contract does not prescribe one. diff --git a/doc/rfc/index.md b/doc/rfc/index.md index 2a23a60c4..8df334478 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -8,6 +8,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting - [Message Queue Tenant Sharding](messagequeue-tenant-sharding.md) - Per-tenant shard key on the platform MySQL message queue; SubmitQueue maps `queueName` to `tenant` at wiring - [Message Queue Contract](messagequeue-contract.md) - How queue payloads are defined (Protobuf, serialized as protobuf JSON), located by audience (external in `api/{domain}/messagequeue/`, internal in `{domain}/core/messagequeue/`), bound to topics (the `topics` proto option), and enforced by Bazel visibility - [Consumer Gate](consumer-gate.md) - Stopping and starting individual queue controllers at runtime via a consumer-side check: blocked deliveries are recorded as parked and postponed back to the queue (re-checked on redelivery), gate state as a separate extension with a file-based first implementation shared by tests and operators +- [Admission Gates](admission-gate.md) - Shared, typed contract for queue-scoped logical admission decisions, with Stovepipe's build gate as the first implementation - [Consumer Hold](consumer-hold.md) - Fourth delivery outcome letting a controller postpone its delivery: the message becomes a partition barrier that pauses consumption for a chosen delay, redelivers in order, and does not count as a failure toward dead-lettering - [Change URIs](change-uri.md) - Identity of a code change: `scheme://{host[:port]}/{path}` per provider (GitHub PR, Phabricator Diff, git ref/commit) and canonical-form rules - [Hooks Framework](hook-framework.md) - Fire-and-forget side effects off pipeline lifecycle events: one shared `HookEvent` contract (`api/base/hook/`) published to a durable per-domain hook topic, dispatched by a per-domain stage to a pluggable hook extension (`platform/extension/hook/`) for integrations like warehouse export and code-review notifications @@ -27,7 +28,6 @@ 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 deleted file mode 100644 index 3ac657947..000000000 --- a/doc/rfc/stovepipe/admission-gate.md +++ /dev/null @@ -1,198 +0,0 @@ -# Stovepipe Admission Gates - -Status: proposed. This RFC defines the framework and extension contract; storage and controller integration land separately. - -## Problem - -Stovepipe currently implements one build-admission policy directly in `process`: admit only while the queue's in-flight count is below its concurrency limit. New requirements, such as general admission throttling and a cooldown after a failed build, will continue to surface, indicating a need for a standard admission gating framework that individual controllers can use to gate work based on customized policy decisions. - -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 boundaries other than build admission. - -## Scope - -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. - -## Contract - -The vendor-neutral contract lives at `stovepipe/extension/admissiongate`: - -```go -type Blocker string - -type Result struct { - Decision Decision - BlockedBy []Blocker -} - -type Gate interface { - TryAdmit(context.Context, entity.Request) (Result, error) -} - -type Config struct { - QueueName string -} - -type Gates interface { - For(Config) (Gate, error) -} -``` - -`Gates` is the host-owned resolver across queues for one logical boundary. The controller receives the resolver for the admission it performs. Like the `buildrunner`, `sourcecontrol`, and `storage` resolver contracts, `For` takes a typed `Config` containing only `QueueName`; wiring uses that identity to select and bind the implementation. 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 for request-stage decisions. Passing only its string ID would discard the queue and immutable request identity already available to the controller, force implementations to parse an ID or add another lookup merely to recover that identity, and diverge from other decision extensions that receive the stage entity. The request is a reference, not a bundle of pre-resolved policy facts: a gate reloads mutable state and resolves queue-scoped storage, configuration, request history, clocks, or remote services through dependencies injected when its implementation is constructed. - -`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 typed `Blocker` values: stable, low-cardinality policy identifiers for logs and metrics, not human-readable errors. Each implementation defines constants for the policies it can report. -- `DecisionUnknown` is not a runtime outcome. It is the invalid zero value, consistent with entity enums elsewhere in the repository, so an implementation that accidentally returns an empty `Result` fails closed. The controller converts a nil-error result with this decision into an error. - -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. - -## Current Behavior - -On `main`, `process` loads the queue and its `QueueConfig`, coalesces the request against `Queue.LatestRequestID`, and compares `Queue.InFlightCount` with `QueueConfig.MaxConcurrent`. When the queue is full, it calls `delivery.Hold(QueueConfig.GateWaitDelayMs)` and returns successfully. `GateWaitDelayMs` is only the delay before redelivery and another concurrency check; it does not impose a minimum interval between admitted builds. Redelivery starts from request loading and coalescing, so a newer head can supersede the deferred request. - -When capacity is available, `process` claims it by incrementing `Queue.InFlightCount` with a queue-version CAS. A queue version conflict reloads the queue and repeats coalescing before another claim. After a terminal runner result, `buildsignal` decrements the counter before marking the request terminal. Relevant DLQ paths also decrement it so abandoned processing work does not permanently consume capacity. There is no minimum-admission timestamp, failure-cooldown state, or opaque admission payload on the queue today. - -## Process Integration - -`process` retains responsibility for request choreography; the gate owns only admission policy and its reservation. The integration preserves the current re-check behavior: - -1. Load the request and queue, then coalesce it against the latest request ID. -2. Resolve the gate with `gates.For(admissiongate.Config{QueueName: request.Queue})`. -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. - -## State And Storage - -The extension contract does not prescribe storage. A stateless gate stores nothing; another implementation may use an implementation-owned table, a key-value store, or a remote quota service. Those dependencies are injected when the implementation is constructed, and neither `Gates` nor `Gate` exposes a generic state API. - -The proposed standard composite build gate does need a small amount of durable state for idempotent reservations, concurrency reconciliation, minimum-interval history, and failure cooldown. Its 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. - -Keeping this implementation's state on the queue is deliberate rather than a framework requirement. The standard build gate must order its reservation with `Queue.LatestRequestID`, matching the current queue CAS that prevents a newly superseded head from claiming capacity. A separate table would give admission state its own CAS but could not atomically observe the latest-head update without a cross-entity transaction. A different gate whose facts do not need that ordering should own its own table or backend instead of adding data to this payload. - -The standard gate exclusively owns the payload's encoding and compatibility. Its 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. Another gate does not read or write this envelope. Changing the standard implementation for a live queue requires that the replacement understand or explicitly migrate the prior payload. - -A representative initial payload is: - -```json -{ - "version": 1, - "active_request_ids": [ - "request/monorepo/main/42" - ], - "policies": { - "minimum_interval": { - "last_admitted_at_ms": 1789506000000 - }, - "failure_cooldown": { - "not_before_ms": 1789509600000 - } - } -} -``` - -This shape illustrates the minimum facts the initial policies need, not a shared wire contract. Active request IDs support idempotency and concurrency reconciliation; the last admission time survives completion for general throttling; and the cooldown deadline survives removal of the failed request. 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 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. - -This reconciliation replaces the current shared-counter ownership in which `process` increments `Queue.InFlightCount` and `buildsignal` or a DLQ path decrements it. Under the proposal, `buildsignal` remains policy-neutral: it persists the build result, request terminal state, and request log, but neither understands the gate envelope nor invokes an admission callback. The next `TryAdmit` observes those durable facts and removes completed reservations. - -## Policy Composition - -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: - -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:** preserve the current rule by deferring while the number of reconciled active admissions is at the per-queue limit. -- **Minimum interval:** add general throttling by requiring at least the configured number of milliseconds between admission timestamps. Non-positive values disable it. -- **Failure cooldown:** add outcome-sensitive throttling by deferring until a runner-reported failure's occurrence time plus the configured cooldown. Non-positive values disable it. - -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 - -Today `queueconfig` supplies `MaxConcurrent` and `GateWaitDelayMs`, and the service uses its built-in default implementation. The proposed minimum-interval and failure-cooldown settings belong in the same typed queue configuration contract; a deployment-backed configuration implementation is separate implementation work. Mutable observations such as admission times and failure facts belong 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 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. 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; 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. - -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 a concurrency policy matching the current `InFlightCount < MaxConcurrent` behavior. -3. Wire `process` through `Gates`, replace its direct counter claim, and make reconciliation authoritative instead of direct releases from `buildsignal` and DLQ paths. -4. Add minimum interval as a new standard policy for general admission throttling. -5. Add failure cooldown as another new 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/platform/extension/admissiongate/BUILD.bazel similarity index 55% rename from stovepipe/extension/admissiongate/BUILD.bazel rename to platform/extension/admissiongate/BUILD.bazel index 9a28cf3b4..8b88d1f54 100644 --- a/stovepipe/extension/admissiongate/BUILD.bazel +++ b/platform/extension/admissiongate/BUILD.bazel @@ -3,7 +3,6 @@ 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", + importpath = "github.com/uber/submitqueue/platform/extension/admissiongate", visibility = ["//visibility:public"], - deps = ["//stovepipe/entity:go_default_library"], ) diff --git a/platform/extension/admissiongate/README.md b/platform/extension/admissiongate/README.md new file mode 100644 index 000000000..ddbe6b669 --- /dev/null +++ b/platform/extension/admissiongate/README.md @@ -0,0 +1,5 @@ +# Admission Gate + +Vendor-neutral contract for deciding whether a domain entity may cross a logical pipeline boundary. See the [Admission Gates RFC](../../../doc/rfc/admission-gate.md) for decision semantics, policy ownership, composition, and controller integration. + +The generic entity parameter preserves each domain's stage-level input contract: Stovepipe binds it to `entity.Request`, while a future SubmitQueue use may bind it to `entity.Batch` or another domain entity. Implementations evaluate policy through injected dependencies and need not own storage. A controller receives a `Gates[T]` resolver for its boundary, and that resolver selects one composite gate from a queue-scoped `Config`; concrete queue routing belongs in service wiring. diff --git a/stovepipe/extension/admissiongate/admissiongate.go b/platform/extension/admissiongate/admissiongate.go similarity index 59% rename from stovepipe/extension/admissiongate/admissiongate.go rename to platform/extension/admissiongate/admissiongate.go index eebc72589..d8f5af4ae 100644 --- a/stovepipe/extension/admissiongate/admissiongate.go +++ b/platform/extension/admissiongate/admissiongate.go @@ -12,17 +12,13 @@ // 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 boundary. +// Package admissiongate defines the shared extension contract for deciding +// whether a domain entity may cross a logical pipeline boundary. package admissiongate //go:generate mockgen -source=admissiongate.go -destination=mock/admissiongate_mock.go -package=mock -import ( - "context" - - "github.com/uber/submitqueue/stovepipe/entity" -) +import "context" // Decision is the expected outcome of evaluating an admission gate. type Decision string @@ -30,7 +26,8 @@ type Decision string const ( // DecisionUnknown is the invalid zero value. DecisionUnknown Decision = "" - // DecisionAdmitted means the request's admission is durably reserved. + // DecisionAdmitted means current policy permits the controller to continue + // admitting the candidate. DecisionAdmitted Decision = "admitted" // DecisionDeferred means policy currently prevents admission. DecisionDeferred Decision = "deferred" @@ -41,20 +38,20 @@ type Blocker string // Result describes an expected admission outcome. type Result struct { - // Decision is whether the request was admitted or deferred. + // Decision is whether the candidate was admitted or deferred. Decision Decision // BlockedBy contains stable policy identifiers when Decision is deferred. BlockedBy []Blocker } -// 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 { - // TryAdmit evaluates current policy and durably reserves an allowed - // admission before returning DecisionAdmitted. A policy denial returns +// Gate decides whether candidates may cross one queue-scoped pipeline +// boundary. T is the owning domain's entity at that pipeline stage. +// Implementations resolve the policies and facts they need from the +// candidate's identity. +type Gate[T any] interface { + // TryAdmit evaluates current policy. A policy denial returns // DecisionDeferred rather than an error. - TryAdmit(ctx context.Context, request entity.Request) (Result, error) + TryAdmit(ctx context.Context, candidate T) (Result, error) } // Config identifies the queue for which a Gate is resolved. @@ -63,10 +60,11 @@ type Config struct { QueueName string } -// Gates resolves the gate for a queue. 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 { +// Gates resolves the gate for a queue and domain entity type. 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[T any] interface { // For returns the Gate selected for config. - For(config Config) (Gate, error) + For(config Config) (Gate[T], error) } diff --git a/stovepipe/extension/admissiongate/mock/BUILD.bazel b/platform/extension/admissiongate/mock/BUILD.bazel similarity index 55% rename from stovepipe/extension/admissiongate/mock/BUILD.bazel rename to platform/extension/admissiongate/mock/BUILD.bazel index 9ca4569b0..c0d657faf 100644 --- a/stovepipe/extension/admissiongate/mock/BUILD.bazel +++ b/platform/extension/admissiongate/mock/BUILD.bazel @@ -3,11 +3,10 @@ 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", + importpath = "github.com/uber/submitqueue/platform/extension/admissiongate/mock", visibility = ["//visibility:public"], deps = [ - "//stovepipe/entity:go_default_library", - "//stovepipe/extension/admissiongate:go_default_library", + "//platform/extension/admissiongate:go_default_library", "@org_uber_go_mock//gomock:go_default_library", ], ) diff --git a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go b/platform/extension/admissiongate/mock/admissiongate_mock.go similarity index 54% rename from stovepipe/extension/admissiongate/mock/admissiongate_mock.go rename to platform/extension/admissiongate/mock/admissiongate_mock.go index 133ba8ef3..208a4ecc4 100644 --- a/stovepipe/extension/admissiongate/mock/admissiongate_mock.go +++ b/platform/extension/admissiongate/mock/admissiongate_mock.go @@ -13,85 +13,84 @@ import ( context "context" reflect "reflect" - entity "github.com/uber/submitqueue/stovepipe/entity" - admissiongate "github.com/uber/submitqueue/stovepipe/extension/admissiongate" + admissiongate "github.com/uber/submitqueue/platform/extension/admissiongate" gomock "go.uber.org/mock/gomock" ) // MockGate is a mock of Gate interface. -type MockGate struct { +type MockGate[T any] struct { ctrl *gomock.Controller - recorder *MockGateMockRecorder + recorder *MockGateMockRecorder[T] isgomock struct{} } // MockGateMockRecorder is the mock recorder for MockGate. -type MockGateMockRecorder struct { - mock *MockGate +type MockGateMockRecorder[T any] struct { + mock *MockGate[T] } // NewMockGate creates a new mock instance. -func NewMockGate(ctrl *gomock.Controller) *MockGate { - mock := &MockGate{ctrl: ctrl} - mock.recorder = &MockGateMockRecorder{mock} +func NewMockGate[T any](ctrl *gomock.Controller) *MockGate[T] { + mock := &MockGate[T]{ctrl: ctrl} + mock.recorder = &MockGateMockRecorder[T]{mock} return mock } // EXPECT returns an object that allows the caller to indicate expected use. -func (m *MockGate) EXPECT() *MockGateMockRecorder { +func (m *MockGate[T]) EXPECT() *MockGateMockRecorder[T] { return m.recorder } // TryAdmit mocks base method. -func (m *MockGate) TryAdmit(ctx context.Context, request entity.Request) (admissiongate.Result, error) { +func (m *MockGate[T]) TryAdmit(ctx context.Context, candidate T) (admissiongate.Result, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "TryAdmit", ctx, request) + ret := m.ctrl.Call(m, "TryAdmit", ctx, candidate) 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 { +func (mr *MockGateMockRecorder[T]) TryAdmit(ctx, candidate any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TryAdmit", reflect.TypeOf((*MockGate)(nil).TryAdmit), ctx, request) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TryAdmit", reflect.TypeOf((*MockGate[T])(nil).TryAdmit), ctx, candidate) } // MockGates is a mock of Gates interface. -type MockGates struct { +type MockGates[T any] struct { ctrl *gomock.Controller - recorder *MockGatesMockRecorder + recorder *MockGatesMockRecorder[T] isgomock struct{} } // MockGatesMockRecorder is the mock recorder for MockGates. -type MockGatesMockRecorder struct { - mock *MockGates +type MockGatesMockRecorder[T any] struct { + mock *MockGates[T] } // NewMockGates creates a new mock instance. -func NewMockGates(ctrl *gomock.Controller) *MockGates { - mock := &MockGates{ctrl: ctrl} - mock.recorder = &MockGatesMockRecorder{mock} +func NewMockGates[T any](ctrl *gomock.Controller) *MockGates[T] { + mock := &MockGates[T]{ctrl: ctrl} + mock.recorder = &MockGatesMockRecorder[T]{mock} return mock } // EXPECT returns an object that allows the caller to indicate expected use. -func (m *MockGates) EXPECT() *MockGatesMockRecorder { +func (m *MockGates[T]) EXPECT() *MockGatesMockRecorder[T] { return m.recorder } // For mocks base method. -func (m *MockGates) For(config admissiongate.Config) (admissiongate.Gate, error) { +func (m *MockGates[T]) For(config admissiongate.Config) (admissiongate.Gate[T], error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "For", config) - ret0, _ := ret[0].(admissiongate.Gate) + ret0, _ := ret[0].(admissiongate.Gate[T]) ret1, _ := ret[1].(error) return ret0, ret1 } // For indicates an expected call of For. -func (mr *MockGatesMockRecorder) For(config any) *gomock.Call { +func (mr *MockGatesMockRecorder[T]) For(config any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates)(nil).For), config) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockGates[T])(nil).For), config) } diff --git a/stovepipe/extension/admissiongate/README.md b/stovepipe/extension/admissiongate/README.md deleted file mode 100644 index 6034a2eaa..000000000 --- a/stovepipe/extension/admissiongate/README.md +++ /dev/null @@ -1,5 +0,0 @@ -# Admission Gate Extension - -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. A controller receives a `Gates` resolver for its boundary, and that resolver selects one composite gate from a queue-scoped `Config`; concrete queue routing belongs in service wiring.